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.QueueProvider{},  // 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.QueueProvider{
    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_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

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 — worker pool manager (needed by tasker:work)
• Web server — dashboard UI + API at RoutePrefix

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
• 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)
  • func (cfg *Config) AutoFillDatabase(c config.M)
• type QueueProvider
  • func (p *QueueProvider) AddCommands() []app.Command
  • func (p *QueueProvider) AddPublishables() []*app.Publishable
  • func (p *QueueProvider) AddRoutes() app.RouteCallback
  • func (p *QueueProvider) Provide(a app.App) error
  • func (p *QueueProvider) Shutdown(ctx context.Context) error

Constants

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

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

	RedisAddr   string
	RedisPass   string
	RedisDB     int
	RedisPrefix string

	DefaultQueue       string
	DefaultMaxAttempts int

	Workers         map[string]int
	EnableAutoscale bool

	HeartbeatInterval int
	RequeueTimeout    int
	PruneAfterHours   int
}

func DefaultConfig

func DefaultConfig() *Config

func (*Config) ApplyOverrides

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

ApplyOverrides reads from a config.M (the "tasker" section) and overrides matching fields on the Config. Keys are flat (no "tasker." prefix).

func (*Config) AutoFillDatabase

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

AutoFillDatabase reads the app's default SQL connection config and fills DSN/DriverName. Only applies when Driver is "sql" and DSN is empty.

type QueueProvider

type QueueProvider struct {
	Config *Config
}

func (*QueueProvider) AddCommands

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

func (*QueueProvider) AddPublishables

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

func (*QueueProvider) AddRoutes

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

func (*QueueProvider) Provide

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

func (*QueueProvider) Shutdown

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