Background Jobs

┌──────────────────┐     ┌─────────┐     ┌──────────────────┐
│  API Handler     │     │  Redis  │     │  Worker          │
│                  │     │         │     │                  │
│  client.Enqueue  │────>│  Queue  │────>│  handleEmailSend │
│  SendEmail(...)  │     │         │     │  handleImage...  │
│                  │     │         │     │  handleCleanup   │
└──────────────────┘     └─────────┘     └──────────────────┘

// Task type constants
const (
    TypeEmailSend     = "email:send"
    TypeImageProcess  = "image:process"
    TypeTokensCleanup = "tokens:cleanup"
)

// Client wraps asynq.Client for enqueuing background jobs.
type Client struct {
    client *asynq.Client
}

// NewClient creates a new job queue client connected to Redis.
func NewClient(redisURL string) (*Client, error)

// Close shuts down the client connection.
func (c *Client) Close() error

// Every enqueue method takes a context.Context first and accepts
// optional ...EnqueueOption (queue, max retries, delay, idempotency key).

// EnqueueSendEmail enqueues an email send job.
func (c *Client) EnqueueSendEmail(
    ctx context.Context,
    to, subject, template string,
    data map[string]interface{},
    opts ...EnqueueOption,
) error

// EnqueueProcessImage enqueues an image processing job.
// (uploadID is the Upload's UUID string.)
func (c *Client) EnqueueProcessImage(
    ctx context.Context,
    uploadID string,
    key, mimeType string,
    opts ...EnqueueOption,
) error

// EnqueueTokensCleanup enqueues a token cleanup job.
func (c *Client) EnqueueTokensCleanup(ctx context.Context) error

// EmailPayload holds the data for an email send job.
type EmailPayload struct {
    To       string                 `json:"to"`
    Subject  string                 `json:"subject"`
    Template string                 `json:"template"`
    Data     map[string]interface{} `json:"data"`
}

// ImagePayload holds the data for an image processing job.
type ImagePayload struct {
    UploadID string `json:"upload_id"` // Upload UUID
    Key      string `json:"key"`
    MimeType string `json:"mime_type"`
}

// WorkerDeps holds dependencies needed by job handlers.
type WorkerDeps struct {
    DB      *gorm.DB
    Mailer  *mail.Mailer
    Storage *storage.Storage
    Cache   *cache.Cache
}

// StartWorker starts the asynq worker server in a goroutine.
// Returns a stop function and any startup error.
func StartWorker(redisURL string, deps WorkerDeps) (func(), error) {
    redisOpt, _ := asynq.ParseRedisURI(redisURL)

    srv := asynq.NewServer(redisOpt, asynq.Config{
        Concurrency: 10,
        Queues: map[string]int{
            "default":  6,
            "critical": 3,
            "low":      1,
        },
    })

    mux := asynq.NewServeMux()
    mux.HandleFunc(TypeEmailSend, handleEmailSend(deps))
    mux.HandleFunc(TypeImageProcess, handleImageProcess(deps))
    mux.HandleFunc(TypeTokensCleanup, handleTokensCleanup(deps))

    go func() {
        srv.Run(mux)
    }()

    return func() { srv.Shutdown() }, nil
}

// Start the background worker
stopWorker, err := jobs.StartWorker(cfg.RedisURL, jobs.WorkerDeps{
    DB:      db,
    Mailer:  mailer,
    Storage: store,
    Cache:   cacheService,
})
if err != nil {
    log.Fatalf("Failed to start worker: %v", err)
}
defer stopWorker()

// c.Enqueue(ctx, taskType, payload, ...EnqueueOption) wraps asynq and
// applies framework defaults; override per call with an EnqueueOption.

// Email: 3 attempts (important to deliver)
c.Enqueue(ctx, TypeEmailSend, payload, EnqueueOption{MaxRetries: 3})

// Image: 2 attempts (can be re-triggered)
c.Enqueue(ctx, TypeImageProcess, payload, EnqueueOption{MaxRetries: 2})

// Cleanup: 1 attempt (runs hourly anyway)
c.Enqueue(ctx, TypeTokensCleanup, nil, EnqueueOption{MaxRetries: 1})

// Custom: specific queue, timeout, and idempotency key
c.Enqueue(ctx, "invoice:generate", payload, EnqueueOption{
    MaxRetries:     5,
    Queue:          "critical",       // default | critical | low
    Timeout:        30 * time.Second,
    IdempotencyKey: "invoice-" + orderID, // dedupes re-enqueues
})

// 1. Add task type constant
const TypeInvoiceGenerate = "invoice:generate"

// 2. Define payload
type InvoicePayload struct {
    OrderID   string `json:"order_id"` // Order UUID
    UserEmail string `json:"user_email"`
}

// 3. Add enqueue method to Client (Enqueue marshals the payload for you)
func (c *Client) EnqueueGenerateInvoice(ctx context.Context, orderID, email string) error {
    return c.Enqueue(ctx, TypeInvoiceGenerate, InvoicePayload{
        OrderID:   orderID,
        UserEmail: email,
    }, EnqueueOption{MaxRetries: 3})
}

// 4. Write handler function
func handleInvoiceGenerate(deps WorkerDeps) func(ctx context.Context, task *asynq.Task) error {
    return func(ctx context.Context, task *asynq.Task) error {
        var payload InvoicePayload
        if err := json.Unmarshal(task.Payload(), &payload); err != nil {
            return err
        }
        // Generate PDF, send email, etc.
        return nil
    }
}

// 5. Register in worker mux (in StartWorker)
mux.HandleFunc(TypeInvoiceGenerate, handleInvoiceGenerate(deps))

// The nightly task, scheduled with --cron, fans out one task per account.
for _, id := range accountIDs {
    err := client.EnqueueReconcileAccount(ctx, ReconcileAccountPayload{AccountID: id, Day: day},
        EnqueueOption{IdempotencyKey: "reconcile:" + id + ":" + day})
    if errors.Is(err, ErrDuplicateTask) {
        continue // already queued for this account and day
    }
    if err != nil {
        return err // retried with backoff; the keys make the retry safe
    }
}