Workqueue Modules Reference

This skill provides detailed usage instructions for all packages in github.com/weedbox/workqueue-modules — a single workqueue.WorkQueue interface with interchangeable backends, so application code stays backend-agnostic.

Module Index

Shared Interface

Backends

Testing

⚠️ Critical Rules

Source of Truth — Do NOT Fetch from GitHub

This file and its sub-files (modules/*.md) are the complete reference for weedbox/workqueue-modules. DO NOT browse github.com/weedbox/workqueue-modules to look up source code.

Inject the Interface, NOT a Concrete Backend

Application code MUST depend on workqueue.WorkQueue, never on a backend type:

// ✅ Correct - use interface
type Params struct {
    fx.In
    WorkQueue workqueue.WorkQueue
}

// ❌ Wrong - using concrete implementation
type Params struct {
    fx.In
    WorkQueue *gorm_workqueue.GormWorkQueue
}

Single-Load Needs No name Tag; Multi-Load Does

Backends register via fxmodule.InterfaceModule[workqueue.WorkQueue], so the single-load / multi-load rules match common-modules connectors:

// Single backend loaded → no name tag
WorkQueue workqueue.WorkQueue

// Multiple backends loaded side by side → disambiguate by scope
Jobs   workqueue.WorkQueue `name:"jobs"`   // gorm_workqueue.Module("jobs")
Events workqueue.WorkQueue `name:"events"` // nats_workqueue.Module("events")

In tests running multiple fx.App instances in one process, call fxmodule.ResetClaim[workqueue.WorkQueue]() between apps.

Handlers Must Be Idempotent

Every backend delivers at-least-once. Crash recovery (expired leases, unacked JetStream messages) can redeliver a task whose handler already ran. Design handlers so a duplicate execution is harmless.

Quick Start

go get github.com/weedbox/workqueue-modules@latest

Load one backend module and inject the interface:

import (
    "github.com/weedbox/common-modules/postgres_connector"
    "github.com/weedbox/workqueue-modules/postgres_workqueue"
    "github.com/weedbox/workqueue-modules/workqueue"
)

func loadModules() ([]fx.Option, error) {
    return []fx.Option{
        postgres_connector.Module("database"),
        postgres_workqueue.Module("workqueue"), // Phase 2, after its connector
    }, nil
}

Enqueue and consume:

task, err := p.WorkQueue.Enqueue(ctx, "emails", payload,
    workqueue.WithDelay(30*time.Second),
    workqueue.WithMaxRetries(5),
    workqueue.WithMetadata(map[string]string{"tenant": "acme"}),
)

sub, err := p.WorkQueue.Consume(ctx, "emails",
    func(ctx context.Context, task *workqueue.Task) error {
        return send(task.Payload)
    },
    workqueue.WithConcurrency(4),
)
defer sub.Stop(context.Background())

Loading order: backend modules belong in Phase 2 (loadModules) and must be declared after their backing connector (postgres_connector / sqlite_connector for the database backends, nats_connector for nats_workqueue). memory_workqueue has no dependency.

Choosing a Backend

Decision shortcuts:

• Multiple replicas in production → any backend except memory_workqueue; gorm_workqueue, postgres_workqueue, and nats_workqueue all support competing consumers across processes (see the multi-instance rule)
• SQLite / single process → gorm_workqueue (local enqueues wake workers immediately; scheduled tasks fire on time)
• Already on PostgreSQL → postgres_workqueue (same table layout as gorm_workqueue; switching later needs no data migration)
• Already running NATS or need high volume → nats_workqueue
• Unit tests → memory_workqueue, or run workqueuetest.Run against your target backend

Shared Semantics (all backends)

• Task.Attempts = number of handler invocations (first delivery = 1); a handler panic counts as a failed attempt
• A task fails permanently after MaxRetries + 1 failed attempts, then moves to the dead-letter queue with LastError and DiedAt set
• Retry delay: RetryBackoff(base, max, attempt) = min(base * 2^(attempt-1), max); defaults 10s base / 10m cap
• ListDeadTasks orders oldest-first with limit/offset paging; RequeueDeadTask resets the attempt counter; unknown IDs return workqueue.ErrTaskNotFound
• Subscription.Stop(ctx) waits for in-flight handlers; Done() closes once fully stopped
• Handlers receive context.Background(), not the Consume call's context

See workqueue.md for the full API reference.

Related Skills

• common-modules - Database and NATS connectors the backends build on
• module-dev - fxmodule.InterfaceModule pattern used by every backend
• project-dev - Project structure and application setup