Initialize swarm

┌─────────────┐     ┌─────────────────────────────┐
│  Load       │────▶│  API Servers (N replicas)   │
│  Balancer   │     └─────────────────────────────┘
└─────────────┘                   │
                    ┌─────────────┴─────────────┐
                    ▼                           ▼
          ┌─────────────────┐         ┌─────────────────┐
          │  DragonflyDB    │         │  Workers (N)    │
          │  (Sharded)      │         │  (Scale these)  │
          └─────────────────┘         └─────────────────┘

# Initialize swarm
docker swarm init
 
# Deploy stack
docker stack deploy -c docker-stack.yml goflow
 
# Scale workers
docker service scale goflow_worker=20

queue := queue.NewShardedQueue(queue.ShardedConfig{
    Shards: []queue.Config{
        {Address: "redis1:6379"},
        {Address: "redis2:6379"},
        {Address: "redis3:6379"},
        {Address: "redis4:6379"},
    },
    Strategy: queue.LeastLoadedShard,
})

// Worker 1 handles shards 0-1
worker1 := queue.NewPartitionedWorker(shardedQueue, 0)
worker1.Start(ctx, 5)
 
// Worker 2 handles shards 2-3
worker2 := queue.NewPartitionedWorker(shardedQueue, 2)
worker2.Start(ctx, 5)

locker := queue.NewDistributedLock(redisClient)
 
lock, err := locker.TryAcquire(ctx, "job:"+jobID, 
    30*time.Second,  // TTL
    5*time.Second,   // Max wait
)
if err != nil {
    return // Lock held by another worker
}
defer lock.Release(ctx)

sem := queue.NewSemaphore(redisClient, "api_calls", 100)
 
slot, err := sem.Acquire(ctx, 30*time.Second)
if err != nil {
    return errors.New("rate limited")
}
defer sem.Release(ctx, slot)
 
callExternalAPI()

import "github.com/nuulab/goflow/pkg/metrics"
 
// Expose Prometheus endpoint
http.Handle("/metrics", metrics.DefaultMetrics.Handler())
 
// Custom metrics
metrics.JobCompleted()
metrics.JobFailed()
metrics.ObserveJobDuration(start)
metrics.SetQueueDepth(depth)

store := queue.NewEventStore(redisClient)
 
// All events are automatically recorded
// Query job history
events, _ := store.GetJobEvents(ctx, jobID)
 
// Real-time streaming
store.Subscribe(ctx, func(e queue.Event) {
    log.Printf("%s: %s", e.Type, e.JobID)
})

dlq := queue.NewDLQ(redisClient, "main")
 
// Slack alerts
dlq.AddAlerter(queue.NewSlackAlerter(webhookURL, "#alerts"))
 
// Webhook alerts
dlq.AddAlerter(queue.NewWebhookAlerter("https://api.pagerduty.com/..."))
 
// Custom alerts
dlq.AddAlerter(&queue.CallbackAlerter{
    Callback: func(entry queue.DLQEntry) {
        sendEmail("Job failed: " + entry.Job.ID)
    },
})