Queue

Background job processing (Laravel-inspired). Queue is core—no plugin needed. Call queue.Boot() in your app bootstrap.

Recommended reading order: Getting Started → Queue → Horizon → Deployment.

Installation

Queue is initialized in bin/server.go when creating a new app:

import "github.com/CodeSyncr/nimbus/queue"

queue.Boot(&queue.BootConfig{RegisterJobs: start.RegisterQueueJobs})

Configuration

Drivers: sync (runs immediately, dev), redis, database, sqs, kafka.

Defining jobs

type SendEmail struct {
    UserID  int
    Subject string
}

func (j *SendEmail) Handle(ctx context.Context) error {
    // Send email...
    return nil
}

// Optional: cleanup when job fails permanently
func (j *SendEmail) Failed(ctx context.Context, err error) {
    log.Printf("SendEmail failed: %v", err)
}

Dispatching

queue.Dispatch(&jobs.SendEmail{UserID: 12, Subject: "Welcome"}).Dispatch(ctx)
queue.Dispatch(&jobs.SendEmail{...}).Delay(5 * time.Minute).Dispatch(ctx)
queue.Dispatch(&jobs.Report{}).OnQueue("reports").Dispatch(ctx)

Registering jobs

// start/jobs.go
func RegisterQueueJobs() {
    queue.Register(&jobs.SendEmail{})
    queue.Register(&jobs.ProcessVideo{})
}

Running the worker

nimbus queue:work

Or from code: queue.RunWorker(ctx, "default")

Production checklist

• Use durable drivers — prefer redis or database in production
• Fail fast on config — set QUEUE_BOOT_STRICT=true in production
• Tune lease timeouts — set QUEUE_REDIS_VISIBILITY_TIMEOUT_SECONDS / QUEUE_DB_LEASE_SECONDS to at least 2x your p95 job runtime
• Keep jobs idempotent — retries and reclaims can re-run the same job
• Bound retries per job — use Retries(n) and implement Failed()
• Watch retry/reclaim signals — sustained spikes usually indicate worker crashes or flaky dependencies

Queue metrics

When Horizon is enabled, queue metrics are exposed at:

GET /horizon/api/metrics/prometheus

Key counters: nimbus_queue_jobs_dispatched_total, nimbus_queue_jobs_processed_total, nimbus_queue_jobs_failed_total, nimbus_queue_jobs_retried_total, nimbus_queue_jobs_reclaimed_total.

Real-Life Example: Order Processing Pipeline

// app/jobs/process_order.go
type ProcessOrder struct {
    OrderID uint
}

func (j *ProcessOrder) Handle(ctx context.Context) error {
    var order models.Order
    if err := db.Preload("Items.Product").First(&order, j.OrderID).Error; err != nil {
        return fmt.Errorf("order not found: %w", err)
    }

    // 1. Verify payment
    if err := verifyPayment(order); err != nil {
        return fmt.Errorf("payment verification failed: %w", err)
    }

    // 2. Update stock
    for _, item := range order.Items {
        if err := db.Model(&models.Product{}).
            Where("id = ?", item.ProductID).
            Update("stock", gorm.Expr("stock - ?", item.Quantity)).Error; err != nil {
            return fmt.Errorf("stock update failed: %w", err)
        }
    }

    // 3. Update order status
    db.Model(&order).Update("status", "processing")

    // 4. Chain: dispatch follow-up jobs
    queue.Dispatch(&GenerateShippingLabel{OrderID: order.ID}).Dispatch(ctx)
    queue.Dispatch(&SendOrderConfirmation{
        OrderID: order.ID,
        Email:   order.User.Email,
    }).Dispatch(ctx)

    return nil
}

func (j *ProcessOrder) Failed(ctx context.Context, err error) {
    db.Model(&models.Order{}).Where("id = ?", j.OrderID).Update("status", "failed")
    notifyAdmin("Order processing failed", j.OrderID, err)
}

Real-Life Example: Image Processing

type ResizeImage struct {
    ImageID uint
    Sizes   []int // [100, 300, 600, 1200]
}

func (j *ResizeImage) Handle(ctx context.Context) error {
    var image models.Image
    db.First(&image, j.ImageID)

    for _, size := range j.Sizes {
        resized, err := resize(image.Path, size)
        if err != nil {
            return fmt.Errorf("resize %dpx failed: %w", size, err)
        }
        key := fmt.Sprintf("images/%d/%dpx.jpg", image.ID, size)
        if err := storage.Put(key, resized); err != nil {
            return fmt.Errorf("upload failed: %w", err)
        }
    }
    db.Model(&image).Update("processed", true)
    return nil
}

Horizon (Queue Dashboard)

Install with nimbus plugin:install horizon. Pass RedisURL for failed-job storage and live Redis queue depths on the Pending page (queue.RedisAdapter key layout). Optional HORIZON_QUEUES lists extra queue names to probe.

import (
    "os"
    "github.com/CodeSyncr/nimbus/plugins/horizon"
)

app.Use(horizon.NewWithOptions(horizon.Options{
    RedisURL: os.Getenv("REDIS_URL"),
}))
// Dashboard at /horizon (see HORIZON_PATH)

See Horizon for API routes, Prometheus metrics, and programmatic queue.RedisQueueWorkloads.

Best Practices

• Keep jobs small and focused — One job, one responsibility
• Make jobs idempotent — Running twice should be safe (retries happen)
• Serialize minimal data — Store IDs, not full objects
• Chain jobs for pipelines — Dispatch follow-up jobs from Handle()
• Implement Failed() — Always handle permanent failure gracefully
• Use Redis for production — In-memory is only for development
• Monitor with Horizon — Watch for queue backlogs and failures