# Skill: Queue Management with Bull

## Identity
- **Skill ID**: `queue-management`
- **Domain**: Asynchronous Job Processing, Background Workers
- **Technologies**: Bull, BullMQ, Redis, Job Queues
- **Source Agent**: `backend-app-expert.md`

## When to Load This Skill

Load this skill when working on:
- Translation job processing
- Webhook delivery with retry logic
- Background email sending
- Periodic credit allocation
- Scheduled cleanup tasks
- Job status monitoring
- Queue metrics and monitoring

**File patterns:**
- `api/src/queues/**/*.ts`
- `api/src/workers/**/*.ts`
- `api/src/jobs/**/*.ts`
- `api/src/services/job-processor*.ts`

## Core Patterns

### 1. Queue Setup and Configuration

```typescript
import Queue from 'bull';
import Redis from 'ioredis';

// Redis connection
const redisOptions = {
  host: process.env.REDIS_HOST || 'localhost',
  port: parseInt(process.env.REDIS_PORT || '6379'),
  maxRetriesPerRequest: null,
  enableReadyCheck: false
};

// Create translation job queue
export const translationQueue = new Queue('translation-jobs', {
  redis: redisOptions,
  defaultJobOptions: {
    attempts: 3, // Retry up to 3 times
    backoff: {
      type: 'exponential',
      delay: 2000 // Start with 2 seconds, then 4s, 8s
    },
    removeOnComplete: 100, // Keep last 100 completed jobs
    removeOnFail: false // Keep failed jobs for debugging
  }
});

// Webhook delivery queue
export const webhookQueue = new Queue('webhook-deliveries', {
  redis: redisOptions,
  defaultJobOptions: {
    attempts: 5, // More retries for webhooks
    backoff: {
      type: 'exponential',
      delay: 5000 // Start with 5 seconds
    },
    timeout: 30000 // 30 second timeout per attempt
  }
});

// Credit allocation queue (scheduled)
export const creditQueue = new Queue('credit-allocation', {
  redis: redisOptions,
  defaultJobOptions: {
    attempts: 1, // No retries for scheduled jobs
    removeOnComplete: 10
  }
});

// Graceful shutdown
process.on('SIGTERM', async () => {
  await translationQueue.close();
  await webhookQueue.close();
  await creditQueue.close();
  process.exit(0);
});
```

### 2. Adding Jobs to Queue

```typescript
import { translationQueue } from './queues';
import { z } from 'zod';

// Job data schema
const TranslationJobData = z.object({
  jobId: z.string().uuid(),
  userId: z.string().uuid(),
  sourceLang: z.string().length(2),
  targetLang: z.string().length(2),
  content: z.string(),
  model: z.enum(['4b', '27b']),
  tone: z.enum(['neutral', 'formal', 'casual']),
  callbackUrl: z.string().url().optional(),
  callbackSecret: z.string().optional()
});

type TranslationJobData = z.infer<typeof TranslationJobData>;

// Add job to queue
async function queueTranslationJob(data: TranslationJobData) {
  // Validate data
  const validatedData = TranslationJobData.parse(data);
  
  // Add to queue with priority
  const job = await translationQueue.add(validatedData, {
    jobId: validatedData.jobId, // Custom job ID for deduplication
    priority: validatedData.model === '27b' ? 1 : 2, // Premium gets priority
    timeout: validatedData.model === '27b' ? 600000 : 300000 // 10min vs 5min
  });
  
  return job;
}

// Endpoint to submit job
router.post('/jobs', requireApiKey, async (req: AuthRequest, res) => {
  try {
    const userId = req.user!.userId;
    
    // Check credits first
    const user = await prisma.user.findUnique({
      where: { id: userId },
      include: {
        credit_transactions: {
          orderBy: { created_at: 'desc' },
          take: 1
        }
      }
    });
    
    const currentBalance = user.credit_transactions[0]?.balance_after || 0;
    const estimatedCost = estimateTokens(req.body.content) * getCostPerToken(req.body.model);
    
    if (currentBalance < estimatedCost) {
      return res.status(402).json({ error: 'Insufficient credits' });
    }
    
    // Create job in database
    const job = await prisma.translationJob.create({
      data: {
        user_id: userId,
        client_job_id: req.body.job_id,
        status: 'pending',
        source_lang: req.body.source_lang,
        target_lang: req.body.target_lang,
        model: req.body.model,
        tone: req.body.tone || 'neutral',
        content: req.body.content,
        content_hash: createHash('sha256').update(req.body.content).digest('hex'),
        callback_url: req.body.callback_url,
        callback_secret: req.body.callback_secret
      }
    });
    
    // Add to queue
    await queueTranslationJob({
      jobId: job.id,
      userId: job.user_id,
      sourceLang: job.source_lang,
      targetLang: job.target_lang,
      content: job.content,
      model: job.model as '4b' | '27b',
      tone: job.tone as 'neutral' | 'formal' | 'casual',
      callbackUrl: job.callback_url || undefined,
      callbackSecret: job.callback_secret || undefined
    });
    
    res.status(202).json({
      success: true,
      job_id: job.id,
      client_job_id: job.client_job_id,
      status: 'pending',
      estimated_tokens: estimateTokens(req.body.content)
    });
  } catch (error) {
    res.status(500).json({ error: 'Failed to queue job' });
  }
});
```

### 3. Worker Processing

```typescript
import { Job } from 'bull';
import { translationQueue } from './queues';
import { prisma } from '../lib/prisma';
import axios from 'axios';

interface TranslationResult {
  translation: string;
  tokensUsed: number;
  processingTimeMs: number;
}

// Process translation jobs
translationQueue.process(5, async (job: Job<TranslationJobData>) => {
  const startTime = Date.now();
  
  try {
    // Update job status to processing
    await prisma.translationJob.update({
      where: { id: job.data.jobId },
      data: { status: 'processing' }
    });
    
    // Call Google Gemini API translation service
    const geminiClient = new GeminiClient();
    const response = await geminiClient.translate({
      sourceLang: job.data.sourceLang,
      targetLang: job.data.targetLang,
      content: job.data.content,
      tone: job.data.tone,
      preserveFormatting: true
    });
    
    const processingTime = Date.now() - startTime;
    const { translation, tokensUsed } = response.data;
    
    // Calculate cost
    const costPerToken = job.data.model === '4b' ? 0.0000005 : 0.000002;
    const cost = tokensUsed * costPerToken;
    
    // Update job in database
    await prisma.translationJob.update({
      where: { id: job.data.jobId },
      data: {
        status: 'completed',
        translation,
        tokens_used: tokensUsed,
        cost,
        processing_time_ms: processingTime,
        completed_at: new Date()
      }
    });
    
    // Deduct credits
    const lastTransaction = await prisma.creditTransaction.findFirst({
      where: { user_id: job.data.userId },
      orderBy: { created_at: 'desc' }
    });
    
    const currentBalance = lastTransaction?.balance_after || 0;
    
    await prisma.creditTransaction.create({
      data: {
        user_id: job.data.userId,
        type: 'deduction',
        amount: -tokensUsed,
        balance_after: currentBalance - tokensUsed,
        description: `Translation job ${job.data.jobId}`,
        related_job_id: job.data.jobId
      }
    });
    
    // Queue webhook if callback URL provided
    if (job.data.callbackUrl) {
      await webhookQueue.add({
        jobId: job.data.jobId,
        url: job.data.callbackUrl,
        secret: job.data.callbackSecret,
        payload: {
          job_id: job.data.jobId,
          client_job_id: await getClientJobId(job.data.jobId),
          status: 'completed',
          translation,
          tokens_used: tokensUsed,
          cost_usd: cost
        }
      });
    }
    
    return { success: true, tokensUsed, cost };
    
  } catch (error) {
    // Update job status to failed
    await prisma.translationJob.update({
      where: { id: job.data.jobId },
      data: {
        status: 'failed',
        error_message: error.message
      }
    });
    
    // Refund credits if deducted
    const lastTransaction = await prisma.creditTransaction.findFirst({
      where: {
        user_id: job.data.userId,
        related_job_id: job.data.jobId,
        type: 'deduction'
      }
    });
    
    if (lastTransaction) {
      await prisma.creditTransaction.create({
        data: {
          user_id: job.data.userId,
          type: 'refund',
          amount: Math.abs(lastTransaction.amount),
          balance_after: lastTransaction.balance_after + Math.abs(lastTransaction.amount),
          description: `Refund for failed job ${job.data.jobId}`,
          related_job_id: job.data.jobId
        }
      });
    }
    
    throw error; // Re-throw for Bull retry logic
  }
});

// Helper function
async function getClientJobId(jobId: string): Promise<string | null> {
  const job = await prisma.translationJob.findUnique({
    where: { id: jobId },
    select: { client_job_id: true }
  });
  return job?.client_job_id || null;
}
```

### 4. Webhook Delivery Worker

```typescript
import { Job } from 'bull';
import { webhookQueue } from './queues';
import { prisma } from '../lib/prisma';
import axios from 'axios';
import crypto from 'crypto';

interface WebhookJobData {
  jobId: string;
  url: string;
  secret?: string;
  payload: any;
}

// Process webhook deliveries
webhookQueue.process(10, async (job: Job<WebhookJobData>) => {
  const attemptNumber = job.attemptsMade + 1;
  
  try {
    // Generate HMAC signature
    const timestamp = Date.now().toString();
    const payloadString = JSON.stringify(job.data.payload);
    
    let signature: string | undefined;
    if (job.data.secret) {
      signature = crypto
        .createHmac('sha256', job.data.secret)
        .update(payloadString)
        .digest('hex');
    }
    
    // Send webhook
    const response = await axios.post(job.data.url, job.data.payload, {
      headers: {
        'Content-Type': 'application/json',
        'X-Webhook-Signature': signature || '',
        'X-Webhook-Timestamp': timestamp,
        'User-Agent': 'TranslatePressZone/1.0'
      },
      timeout: 30000,
      validateStatus: (status) => status >= 200 && status < 300
    });
    
    // Log successful delivery
    await prisma.webhookDelivery.create({
      data: {
        job_id: job.data.jobId,
        attempt_number: attemptNumber,
        success: true,
        http_status: response.status,
        response_body: JSON.stringify(response.data).substring(0, 1000)
      }
    });
    
    return { success: true };
    
  } catch (error) {
    const httpStatus = error.response?.status || null;
    const errorMessage = error.message;
    const responseBody = error.response?.data 
      ? JSON.stringify(error.response.data).substring(0, 1000)
      : null;
    
    // Log failed delivery
    await prisma.webhookDelivery.create({
      data: {
        job_id: job.data.jobId,
        attempt_number: attemptNumber,
        success: false,
        http_status: httpStatus,
        response_body: responseBody,
        error_message: errorMessage
      }
    });
    
    // Don't retry on 4xx errors (client errors)
    if (httpStatus && httpStatus >= 400 && httpStatus < 500) {
      console.error(`Webhook failed with client error ${httpStatus}, not retrying`);
      return { success: false, permanent: true };
    }
    
    throw error; // Re-throw for Bull retry logic (5xx errors, network issues)
  }
});
```

### 5. Scheduled Jobs (Cron-style)

```typescript
import { creditQueue } from './queues';

// Monthly credit allocation
creditQueue.add(
  'monthly-credit-allocation',
  {},
  {
    repeat: {
      cron: '0 0 1 * *' // First day of month at midnight
    },
    jobId: 'monthly-credit-allocation' // Prevent duplicates
  }
);

// Process scheduled credit allocation
creditQueue.process('monthly-credit-allocation', async (job) => {
  // Get all active subscriptions
  const subscriptions = await prisma.subscription.findMany({
    where: {
      status: 'active',
      current_period_end: {
        gte: new Date()
      }
    },
    include: {
      user: {
        include: {
          credit_transactions: {
            orderBy: { created_at: 'desc' },
            take: 1
          }
        }
      }
    }
  });
  
  // Allocate credits based on plan
  const planCredits = {
    starter: 100000,      // 100K tokens
    professional: 500000, // 500K tokens
    enterprise: 2000000   // 2M tokens
  };
  
  for (const subscription of subscriptions) {
    const credits = planCredits[subscription.plan_tier];
    const currentBalance = subscription.user.credit_transactions[0]?.balance_after || 0;
    
    await prisma.creditTransaction.create({
      data: {
        user_id: subscription.user_id,
        type: 'allocation',
        amount: credits,
        balance_after: currentBalance + credits,
        description: `Monthly allocation for ${subscription.plan_tier} plan`,
        related_payment_id: null
      }
    });
  }
  
  return { allocated: subscriptions.length };
});

// Daily cleanup of old completed jobs
creditQueue.add(
  'cleanup-old-jobs',
  {},
  {
    repeat: {
      cron: '0 2 * * *' // Daily at 2 AM
    },
    jobId: 'cleanup-old-jobs'
  }
);

creditQueue.process('cleanup-old-jobs', async () => {
  const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
  
  const deleted = await prisma.translationJob.deleteMany({
    where: {
      status: 'completed',
      completed_at: {
        lt: thirtyDaysAgo
      }
    }
  });
  
  return { deleted: deleted.count };
});
```

### 6. Queue Monitoring and Metrics

```typescript
import { translationQueue, webhookQueue } from './queues';

// Admin endpoint for queue metrics
router.get('/admin/queues', requireAuth, async (req, res) => {
  const [translationCounts, webhookCounts] = await Promise.all([
    translationQueue.getJobCounts(),
    webhookQueue.getJobCounts()
  ]);
  
  // Get failed jobs
  const failedTranslations = await translationQueue.getFailed(0, 10);
  const failedWebhooks = await webhookQueue.getFailed(0, 10);
  
  res.json({
    translation_queue: {
      waiting: translationCounts.waiting,
      active: translationCounts.active,
      completed: translationCounts.completed,
      failed: translationCounts.failed,
      delayed: translationCounts.delayed,
      failed_jobs: failedTranslations.map(job => ({
        id: job.id,
        data: job.data,
        error: job.failedReason,
        attempts: job.attemptsMade
      }))
    },
    webhook_queue: {
      waiting: webhookCounts.waiting,
      active: webhookCounts.active,
      completed: webhookCounts.completed,
      failed: webhookCounts.failed,
      failed_jobs: failedWebhooks.map(job => ({
        id: job.id,
        data: job.data,
        error: job.failedReason,
        attempts: job.attemptsMade
      }))
    }
  });
});

// Retry failed job
router.post('/admin/queues/:queue/jobs/:jobId/retry', requireAuth, async (req, res) => {
  const { queue, jobId } = req.params;
  
  const targetQueue = queue === 'translation' ? translationQueue : webhookQueue;
  const job = await targetQueue.getJob(jobId);
  
  if (!job) {
    return res.status(404).json({ error: 'Job not found' });
  }
  
  await job.retry();
  
  res.json({ success: true });
});

// Queue event listeners for monitoring
translationQueue.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed in ${result.processingTimeMs}ms`);
});

translationQueue.on('failed', (job, error) => {
  console.error(`Job ${job.id} failed:`, error.message);
  // Send alert to admin
});

translationQueue.on('stalled', (job) => {
  console.warn(`Job ${job.id} stalled`);
});
```

## Anti-Patterns (Forbidden)

| ❌ Mistake | ✅ Fix |
|-----------|--------|
| Processing jobs in HTTP request handlers | Use queue for async processing |
| Not handling worker crashes gracefully | Implement graceful shutdown and health checks |
| Infinite retries on failed jobs | Set `attempts` limit (3-5 attempts) |
| Not using exponential backoff | Configure `backoff: { type: 'exponential' }` |
| Storing large data in job payload | Store in database, pass ID only |
| Not monitoring queue health | Implement metrics endpoints and alerts |
| Retrying client errors (4xx) | Only retry 5xx and network errors |
| Not cleaning up completed jobs | Set `removeOnComplete` option |
| Blocking Redis connection | Use separate Redis connections for queues |
| Not validating job data | Use Zod schemas for type safety |

## Integration with Other Skills

**Often combined with:**
- `api-endpoint-creation` - Job submission endpoints
- `authentication-security` - Protecting admin queue endpoints
- `ml-service-integration` - Calling translation service
- `webhook-implementation` - Delivering callbacks
- `error-handling-logging` - Logging job failures

**Depends on:**
- `database-schema-design` - TranslationJob, WebhookDelivery tables

## Environment Variables Required

```bash
# Redis Configuration (container hostnames in bridge network)
REDIS_HOST="redis"
REDIS_PORT="6379"
REDIS_PASSWORD=""  # Optional

# Google Gemini API (Translation Engine)
GEMINI_API_KEY="your-google-gemini-api-key"
GEMINI_MODEL="gemini-3-flash-preview"

# Queue Configuration
QUEUE_CONCURRENCY_TRANSLATION=5  # Concurrent translation jobs
QUEUE_CONCURRENCY_WEBHOOK=10     # Concurrent webhook deliveries
```

## Quick Reference

### Queue Configuration Options

```typescript
{
  attempts: 3,              // Retry count
  backoff: {
    type: 'exponential',    // or 'fixed'
    delay: 2000            // Initial delay in ms
  },
  timeout: 300000,          // 5 minutes
  removeOnComplete: 100,    // Keep last 100
  removeOnFail: false,      // Keep failed jobs
  priority: 1,              // Lower = higher priority
  jobId: 'unique-id'        // Custom ID (prevents duplicates)
}
```

### Retry Strategies by Job Type

| Job Type | Attempts | Backoff | Timeout |
|----------|----------|---------|---------|
| Translation (4b) | 3 | Exponential (2s) | 5 minutes |
| Translation (27b) | 3 | Exponential (2s) | 10 minutes |
| Webhook Delivery | 5 | Exponential (5s) | 30 seconds |
| Credit Allocation | 1 | None | 1 minute |
| Cleanup Tasks | 2 | Fixed (60s) | 5 minutes |

### Job Status Lifecycle

```
pending → processing → completed
                    ↘ failed (with retries)
                    ↘ cancelled (manual)
```

### Bull Events

- `completed` - Job finished successfully
- `failed` - Job failed (after all retries)
- `stalled` - Job hasn't reported progress
- `progress` - Job reported progress
- `active` - Job started processing
- `waiting` - Job added to queue
- `removed` - Job removed from queue

## Validation Checklist

Before completing queue implementation:

- [ ] Redis connection configured with retry logic
- [ ] All queues have `defaultJobOptions` set
- [ ] Worker concurrency tuned for system resources
- [ ] Exponential backoff configured for retries
- [ ] Job timeouts prevent hanging workers
- [ ] Failed jobs retained for debugging (`removeOnFail: false`)
- [ ] Completed jobs cleaned up (`removeOnComplete` set)
- [ ] Graceful shutdown implemented (SIGTERM handler)
- [ ] Queue metrics endpoint implemented
- [ ] Job data validated with Zod schemas
- [ ] Client errors (4xx) don't trigger retries
- [ ] Credits refunded on job failure
- [ ] Webhook signatures generated correctly
- [ ] Scheduled jobs use `jobId` to prevent duplicates
- [ ] Queue health monitored (stalled job alerts)

## Worker Deployment

### Development
```bash
# Start API server
npm run dev

# Start worker (separate process)
npm run worker
```

### Production (Podman Compose)
All services defined in `press-zone-backend/podman-compose.yml` with bridge network `tpz-backend`.
Containers use service names as hostnames (`redis`, `postgres`).

```bash
# Deploy updated worker
cd ~/Press.Zone-Works/press-zone-backend && podman-compose up -d --build worker

# View worker logs
podman-compose logs --tail 50 worker
```

### Horizontal Scaling
- Multiple worker instances process from same queue
- Bull handles distributed locking automatically
- Scale workers independently from API servers
- Monitor Redis memory usage

## Common Issues & Solutions

| Issue | Cause | Solution |
|-------|-------|----------|
| Jobs stuck in "active" | Worker crash | Implement stalled job detection |
| Memory growth | Not removing completed jobs | Set `removeOnComplete` option |
| Slow processing | Too many concurrent jobs | Reduce `process()` concurrency |
| Lost jobs | Redis persistence disabled | Enable Redis AOF persistence |
| Duplicate jobs | No `jobId` specified | Use unique `jobId` for idempotency |
| High Redis memory | Large job payloads | Store data in Postgres, pass ID only |