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.

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)
• 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)

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 {
	//	    user, ok := auth.UserAs[*models.User](c)
	//	    return ok && slices.Contains(admins, user.Email)
	//	},
	//
	// Authentication has already run by the time this is called, so a
	// session cookie or a bearer token both resolve the same user. Nothing
	// stops a check that ignores users entirely — a shared secret header,
	// or 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
	// 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