Queue & Background Jobs

Overview

Lemmego includes a built-in background job system (tasker) wrapped as a first-class service provider (queue). It supports multiple queues, worker pools, retries with backoff, scheduled jobs, batch operations, multi-node clustering, and a real-time web dashboard.

Quick Start

The queue provider is already registered in new projects:

// bootstrap/providers.go
&queue.Provider{}

With no config, it auto-reads the database connection from your existing configs/database.go. Jobs are stored in the same database as your app.

Defining a Job

Jobs implement the Job interface:

package jobs

import (
    "context"
    "fmt"
    "github.com/lemmego/queue"
)

type SendWelcomeEmail struct {
    UserID int    `json:"user_id"`
    Email  string `json:"email"`
}

func (j *SendWelcomeEmail) Handle(ctx context.Context) error {
    fmt.Printf("Sending email to %s\n", j.Email)
    return nil
}

func init() {
    queue.RegisterJob("*jobs.SendWelcomeEmail", func() queue.Job {
        return &SendWelcomeEmail{}
    })
}

Dispatching Jobs

queue.Dispatch(ctx, &SendWelcomeEmail{
    UserID: 42,
    Email:  "user@test.com",
})

// With options
queue.Dispatch(ctx, &SendWelcomeEmail{UserID: 43},
    queue.OnQueue("email"),
    queue.WithDelay(5*time.Minute),
    queue.WithTags("priority"),
)

Running Workers

lemmego run tasker:work --queue=default,email --workers=3

Workers claim jobs from the database atomically and process them. Run multiple worker processes across different machines — they coordinate through the shared database.

Optional Job Interfaces

Customize job behavior by implementing optional interfaces:

Interface	Purpose
`ShouldQueue`	Set a custom queue name
`ShouldDelay`	Delay job execution
`ShouldRetry`	Max retry attempts
`ShouldFail`	Called when all retries exhausted
`ShouldTag`	Tags for dashboard filtering
`ShouldBatch`	Group jobs into batches
`ShouldTimeout`	Per-job timeout
`BeforeHandleHook`	Run before job execution
`AfterHandleHook`	Run after successful execution
`BeforeRetryHook`	Run before each retry

Retry & Backoff

By default, jobs use exponential backoff: 2^attempt * 2s, max 24h. Max 3 attempts. Configure per job:

func (j *MyJob) MaxAttempts() int { return 5 }

Failed jobs appear on the dashboard with full error stack traces. Click “Retry” to re-queue them.

Worker Pools

Configure pools per queue:

&queue.Provider{
    Config: &queue.Config{
        Workers: map[string]int{
            "default": 3,
            "email":   5,
            "images":  10,
        },
    },
}

Enable auto-scaling to dynamically adjust workers based on queue backlog:

Config: &queue.Config{
    EnableAutoscale: true,
}

Scheduling Recurring Jobs

scheduler.New(mgr).Register(scheduler.ScheduledJob{
    ID:       "daily-report",
    Schedule: "0 6 * * *",  // every day at 6 AM
    Job:      &GenerateReport{},
    Queue:    "default",
})

Web Dashboard

Visit http://localhost:8080/tasker/ for the real-time dashboard:

Feature	Description
Stat cards	Status, jobs/min, processes, failed count
Queues table	Available/running/completed/failed per queue, pause/resume
Jobs table	Filterable by state (failed, retryable, running, etc.)
Job detail	Payload, errors, execution timeline, output
Actions	Retry, cancel, batch retry, batch cancel
Workers	Connected nodes with uptime
Metrics	Per-job-type throughput, avg runtime, fail rate
Live updates	Auto-refresh every 10 seconds + SSE real-time stats

API Endpoints

All endpoints are mounted under /tasker/:

Method	Path	Description
`GET`	`/tasker/`	Dashboard HTML
`GET`	`/tasker/api/stats`	Global stats
`GET`	`/tasker/api/jobs`	Paginated job list
`GET`	`/tasker/api/jobs/{id}`	Job detail
`POST`	`/tasker/api/jobs/{id}/retry`	Retry a job
`POST`	`/tasker/api/jobs/{id}/cancel`	Cancel a job
`POST`	`/tasker/api/jobs/batch/retry`	Batch retry
`POST`	`/tasker/api/jobs/batch/cancel`	Batch cancel
`GET`	`/tasker/api/queues`	Queue metrics
`POST`	`/tasker/api/queues/{queue}/pause`	Pause a queue
`POST`	`/tasker/api/queues/{queue}/resume`	Resume a queue
`GET`	`/tasker/api/workers`	Connected nodes
`GET`	`/tasker/api/metrics/jobs`	Per-job-type metrics
`GET`	`/tasker/api/events`	SSE real-time stream
`POST`	`/tasker/api/prune`	Prune old jobs

Configuration

Via provider options

&queue.Provider{
    Config: &queue.Config{
        Driver:             "sql",         // or "redis"
        DSN:                "postgres://...",
        DriverName:         "postgres",
        TablePrefix:        "tasker_",
        RoutePrefix:        "/tasker",
        DefaultQueue:       "default",
        DefaultMaxAttempts: 3,
        Workers:            map[string]int{"default": 3},
    },
}

Via .env

TASKER_ROUTE_PREFIX=/jobs
TASKER_TABLE_PREFIX=myapp_
TASKER_QUEUE=default
TASKER_MAX_ATTEMPTS=3
TASKER_HEARTBEAT_SEC=5
TASKER_REQUEUE_SEC=60
TASKER_PRUNE_HOURS=168
TASKER_AUTOSCALE=false

Driver Support

SQL (PostgreSQL / MySQL / SQLite)

Default driver. Stores jobs in a tasker_jobs table with atomic claiming using SELECT ... FOR UPDATE SKIP LOCKED. Supports all SQL databases.

Redis

Alternative driver for node coordination, leader election, and distributed locking (job queue implementation uses database).

Architecture

   ┌──────────────┐     ┌──────────────┐
   │   Web App    │     │  Worker 1    │
   │  Dispatch    │     │  Supervisor  │
   │  Dashboard   │     │  └─Pool (3)  │
   └──────┬───────┘     └──────┬───────┘
          │                    │
          └────────┬───────────┘
                   │
          ┌────────▼────────┐
          │   PostgreSQL    │
          │  (jobs table)   │
          │  SELECT FOR     │
          │  UPDATE SKIP    │
          │  LOCKED         │
          └─────────────────┘

Multiple workers share the same database — they claim available jobs atomically. No direct inter-node communication needed. Scale by adding more worker processes.

CLI Commands

lemmego run tasker:work            # Start worker
  --queue=default,email            #   Queues to listen on
  --workers=3                      #   Workers per queue

lemmego run tasker:dispatch        # Dispatch test jobs
  --count=10
  --queue=default
  --fail-rate=30                   # % chance of failure
  --delay=500                      # Simulated processing delay (ms)

lemmego run tasker:job retry 5     # Retry job #5
lemmego run tasker:job cancel 5    # Cancel job #5

lemmego run tasker:queue list      # List queue stats

Caching CLI Prompting