queue

README

Queue — Lemmego Tasker Plugin

queue is a standalone, self-contained module that wraps tasker as a lemmego-compatible provider. It implements Provider, CommandProvider, RouteProvider, PublishableProvider, and ShutdownProvider interfaces.

Register your driver

This module does not import a database/sql driver, so your application does:

import _ "github.com/lib/pq"              // postgres
import _ "github.com/glebarez/go-sqlite"  // sqlite
import _ "github.com/go-sql-driver/mysql" // mysql

lemmego new writes the one your project chose. If you are adding a database to an existing project, add it yourself — a missing driver fails when the connection opens, with database/sql's own message naming the forgotten import. That is at boot, on every environment, so it is not a mistake you can ship.

It used to import all three, so every project linked the two it had never configured; the pure-Go sqlite driver alone is twenty-nine packages of translated libc.

Usage

In bootstrap/providers.go

import "github.com/lemmego/queue"

func LoadProviders() []app.Provider {
    return []app.Provider{
        &queue.Provider{},  // auto-configures from app config
    }
}

That's it. The provider auto-reads database credentials from your existing configs/database.go (same sql.connections.<default> that gormconnector uses). No hardcoded DSN needed.

With explicit configuration

&queue.Provider{
    Config: &queue.Config{
        Driver:             "redis",
        RedisAddr:          ":6379",
        RoutePrefix:        "/background-jobs",  // default: "/tasker"
        DefaultQueue:       "default",
        DefaultMaxAttempts: 3,
        Workers:            map[string]int{"default": 3, "email": 2},
        EnableAutoscale:    false,
    },
}

Configuration Priority

1. Explicit Config fields (highest)
2. App config (published tasker.go + env vars)
3. Sensible defaults (lowest)

.env overrides

TASKER_ROUTE_PREFIX=/jobs
TASKER_DRIVER=sql
TASKER_DSN=
TASKER_SQL_DRIVER=
TASKER_TABLE_PREFIX=myapp_
TASKER_MAX_OPEN_CONNS=25
TASKER_MAX_IDLE_CONNS=10
TASKER_CONN_MAX_LIFETIME_SEC=0
TASKER_QUEUE=default
TASKER_MAX_ATTEMPTS=3
TASKER_WORKERS=3
TASKER_HEARTBEAT_SEC=5
TASKER_REQUEUE_INTERVAL_SEC=30
TASKER_REQUEUE_SEC=60
TASKER_PRUNE_INTERVAL_HOURS=24
TASKER_PRUNE_HOURS=168
TASKER_AUTOSCALE=false

When Config is nil (auto-mode)

• Reads sql.connections.<default> from app config (same source as gormconnector)
• Picks up driver, database, host, port, user, password
• Works with SQLite, MySQL, PostgreSQL out of the box
• Uses tasker_ table prefix, /tasker route prefix

When Driver is "redis"

Skips SQL config entirely — use RedisAddr, RedisPass, RedisDB fields.

What Gets Created

Always (HTTP + Console)

• Driver — database connection (SQL or Redis)
• Migration — tables auto-created (CREATE TABLE IF NOT EXISTS)
• Manager — job dispatcher, registered as tasker.Global()

HTTP only (skipped in console mode)

• Supervisor — registered for dashboard controls, but not automatically started
• Web server — dashboard UI + API at RoutePrefix

HTTP startup does not implicitly run workers because the application service container has no start lifecycle. Run tasker:work as a separately managed process. The provider stops any active queue resources and closes its database or Redis connection during application shutdown.

Commands

Registered automatically via CommandProvider:

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

Routes

Mounted automatically via RouteProvider:

Prefix	Description
GET /tasker/	Dashboard UI
GET /tasker/api/*	JSON API
POST /tasker/api/*	Actions (retry, cancel, pause, resume, prune)

Customize with Config.RoutePrefix:

Config: &Config{RoutePrefix: "/jobs"}  // → GET /jobs/api/stats

Publishing Config

The provider publishes internal/configs/tasker.go via AddPublishables():

lemmego run publish --tags=config

This creates a config file with env-var-driven defaults for all tasker settings.

Dependencies

• github.com/lemmego/tasker — core job system
• github.com/lemmego/api — lemmego framework interfaces
• github.com/spf13/cobra — CLI commands

No GPA dependency. The driver manages its own connection pool directly.

License

MIT

Documentation

Index

• Constants
• Variables
• func AddGlobalMiddleware(mw tasker.JobMiddleware)
• func Cancel(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)
• func Chain(ctx context.Context, jobs []tasker.Job, opts ...tasker.DispatchOpt) ([]*tasker.JobRow, error)
• func Dispatch(ctx context.Context, job tasker.Job, opts ...tasker.DispatchOpt) (*tasker.JobRow, error)
• func DispatchBatch(ctx context.Context, jobs []tasker.Job, opts ...tasker.DispatchOpt) ([]*tasker.JobRow, error)
• func GetJob(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)
• func RegisterJob(name string, factory tasker.JobFactory)
• func Retry(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)
• func ScheduleCommand(a app.App) *cobra.Command
• func Scheduler(a app.App) *scheduler.Scheduler
• type Config
  • func DefaultConfig() *Config
  • func (cfg *Config) ApplyOverrides(c config.M)
• type Provider
  • func (p *Provider) AddCommands() []app.Command
  • func (p *Provider) AddPublishables() []*app.Publishable
  • func (p *Provider) AddRoutes() app.RouteCallback
  • func (p *Provider) Provide(a app.App) error
  • func (p *Provider) Shutdown(ctx context.Context) error

Constants

const ConfigStub = `` /* 1664-byte string literal not displayed */

const TagConfig = "queue-config"

TagConfig is what `lemmego publish --tags` selects this module's configuration on.

Variables

var ErrNotInitialized = errors.New("queue: provider is not initialized")

Functions

func AddGlobalMiddleware

func AddGlobalMiddleware(mw tasker.JobMiddleware)

func Cancel

func Cancel(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)

func Chain

func Chain(ctx context.Context, jobs []tasker.Job, opts ...tasker.DispatchOpt) ([]*tasker.JobRow, error)

func Dispatch

func Dispatch(ctx context.Context, job tasker.Job, opts ...tasker.DispatchOpt) (*tasker.JobRow, error)

func DispatchBatch

func DispatchBatch(ctx context.Context, jobs []tasker.Job, opts ...tasker.DispatchOpt) ([]*tasker.JobRow, error)

func GetJob

func GetJob(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)

func RegisterJob

func RegisterJob(name string, factory tasker.JobFactory)

func Retry

func Retry(ctx context.Context, id tasker.JobID) (*tasker.JobRow, error)

func ScheduleCommand

func ScheduleCommand(a app.App) *cobra.Command

ScheduleCommand runs the scheduler in the foreground.

One process, deliberately. tasker's scheduler has no leader election, so two of these dispatch every job twice. Run it as a single replica; for jobs where a double dispatch would be harmful anyway — a weekly rollup, an email — take a cache lock inside the job as well, because "exactly one process" is a promise a deployment makes and occasionally breaks.

func Scheduler

func Scheduler(a app.App) *scheduler.Scheduler

Scheduler returns the recurring-job scheduler, or nil if none was configured.

It is registered in the container whenever Provider.Schedule is set, so a handler or a command can inspect what is scheduled — List() is the honest answer to "what runs on a timer here", which is otherwise a grep.

Types

type Config

type Config struct {
	Driver          string
	DSN             string
	DriverName      string
	TablePrefix     string
	RoutePrefix     string
	MaxOpenConns    int
	MaxIdleConns    int
	ConnMaxLifetime time.Duration

	RedisAddr     string
	RedisPass     string
	RedisDB       int
	RedisPoolSize int
	RedisPrefix   string

	DefaultQueue       string
	DefaultMaxAttempts int

	Workers         map[string]int
	EnableAutoscale bool

	HeartbeatInterval int
	RequeueInterval   int
	RequeueTimeout    int
	PruneInterval     int
	PruneAfterHours   int
	// contains filtered or unexported fields
}

func DefaultConfig

func DefaultConfig() *Config

func (*Config) ApplyOverrides

func (cfg *Config) ApplyOverrides(c config.M)

ApplyOverrides applies every present key, including false and zero values.

type Provider

type Provider struct {
	Config *Config

	// DashboardAuth decides whether a request may use the management
	// dashboard. Return true to let it through, false to answer 401.
	//
	// The dashboard can retry, cancel and delete jobs, so it stays closed
	// until this is set: leaving it nil is not "no authentication", it is
	// "no access". That is deliberate — the failure mode of the opposite
	// default is a public button that clears somebody's queue.
	//
	// It takes an app.Context rather than a user, because "may this person
	// use it" is a question about the request, and the answer usually needs
	// the application's own user type:
	//
	//	DashboardAuth: func(c app.Context) bool {
	//	    if err := auth.Check(c); err != nil {
	//	        return false
	//	    }
	//	    user, ok := auth.UserAs[*models.User](c)
	//	    return ok && slices.Contains(admins, user.Email)
	//	},
	//
	// Call auth.Check yourself, and call it first. The dashboard is mounted
	// as a raw http.Handler, so none of the router's middleware runs for it
	// and nothing has looked at the request's cookie by the time this is
	// called. Omit it and the user is always absent, so the answer is always
	// no — an honest failure, but a puzzling one.
	//
	// Check resolves a session cookie and a bearer token alike, so one rule
	// covers both. Nothing stops a check that ignores users entirely — a
	// shared secret header, an IP allowlist — which is why this is a
	// predicate over the request and not over a user interface.
	DashboardAuth func(c app.Context) bool

	// Schedule returns the jobs this application runs on a timer.
	//
	// Registering them is all this does. Nothing is started, because a
	// scheduler started wherever it was constructed would run in every web
	// process and every worker, and an hourly job would fire once per
	// instance — invisibly, from inside any one of them. Start it with
	// `lemmego run tasker:schedule`, as exactly one process.
	//
	//	Schedule: func(a app.App) []scheduler.ScheduledJob {
	//	    return []scheduler.ScheduledJob{
	//	        {ID: "weekly-rollup", Schedule: "5 0 * * 1", Job: &jobs.RollWeek{}},
	//	    }
	//	}
	//
	// A malformed cron expression fails startup rather than producing a job
	// that never runs.
	Schedule func(a app.App) []scheduler.ScheduledJob
	// contains filtered or unexported fields
}

func (*Provider) AddCommands

func (p *Provider) AddCommands() []app.Command

func (*Provider) AddPublishables

func (p *Provider) AddPublishables() []*app.Publishable

func (*Provider) AddRoutes

func (p *Provider) AddRoutes() app.RouteCallback

func (*Provider) Provide

func (p *Provider) Provide(a app.App) error

func (*Provider) Shutdown

func (p *Provider) Shutdown(ctx context.Context) error