Background Job Patterns

Version: 1.0.0 Last Modified: 2025-12-13 Author: Petunia Team Status: Production

Table of Contents

1. Overview
2. Architecture
3. BullMQ Queue System
4. Database Job Pattern
5. Cron Jobs
6. Retry Strategies
7. Error Handling
8. Monitoring and Debugging
9. Best Practices
10. Examples

---

Overview

Petunia uses three complementary approaches for background job processing:

1. BullMQ (Redis-based) - Production-grade job queue for high-throughput async processing
2. Database Jobs - Poll-based job pattern for durability and simple deployment
3. Cron Jobs - Scheduled recurring tasks via Vercel Cron

When to Use Each Pattern

Pattern	Use Cases	Pros	Cons
BullMQ	High-volume, async processing (analytics, notifications, webhooks)	Fast, scalable, rate limiting, priority queues	Requires Redis, complex setup
Database Jobs	Durable, recoverable tasks (review sync, call log ingestion)	Simple, no external deps, durable	Slower, polling overhead
Cron Jobs	Scheduled maintenance (cleanup, settlements, dunning)	Simple, built-in to Vercel	Limited flexibility, no retry logic

---

Architecture

┌─────────────────────────────────────────────────────────────┐
│                    Background Job System                     │
├─────────────────────────────────────────────────────────────┤
│                                                               │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │   BullMQ     │  │  Database    │  │  Vercel      │       │
│  │   Queues     │  │  Job Tables  │  │  Cron        │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
│         │                 │                  │               │
│         ▼                 ▼                  ▼               │
│  ┌──────────────────────────────────────────────────┐       │
│  │         Job Processing Infrastructure            │       │
│  │  - Retry with exponential backoff                │       │
│  │  - Dead letter queues                            │       │
│  │  - Prometheus metrics                            │       │
│  │  - Sentry monitoring                             │       │
│  └──────────────────────────────────────────────────┘       │
│                                                               │
└─────────────────────────────────────────────────────────────┘

Directory Structure

/lib/queue/               # Queue implementations
  ├── bullmq-queue.ts     # BullMQ wrapper with Redis
  └── message-queue.ts    # In-memory fallback queue

/lib/jobs/                # Database job services
  ├── googleReviewSyncJobService.ts
  ├── callLogSyncJobService.ts
  └── yelpReviewSyncJobService.ts

/app/api/cron/            # Cron job endpoints
  ├── google-reviews/
  ├── call-logs/
  ├── lead-settlement/
  ├── dunning/
  ├── message-status/
  ├── orphan-cleanup/
  ├── presence-cleanup/
  ├── usage-warnings/
  └── data-retention/

---

BullMQ Queue System

Overview

BullMQ provides production-grade job queues backed by Redis for high-throughput async processing.

Location: /lib/queue/bullmq-queue.ts

Queue Names

export const QUEUE_NAMES = {
  TTFV_CALCULATION: 'ttfv-calculation',
  ANALYTICS_ROLLUP: 'analytics-rollup',
  MESSAGE_PROCESSING: 'message-processing',
  NOTIFICATION: 'notification',
  COMPETITOR_ANALYSIS: 'competitor-analysis',
  EMAIL_DELIVERY: 'email-delivery',
  SMS_DELIVERY: 'sms-delivery',
  WEBHOOK_DELIVERY: 'webhook-delivery',
} as const;

Configuration

// Default job options
{
  attempts: 3,
  backoff: {
    type: 'exponential',
    delay: 1000, // 1 second base delay
  },
  removeOnComplete: {
    age: 3600,    // Keep completed jobs for 1 hour
    count: 1000,  // Keep at most 1000 completed jobs
  },
  removeOnFail: {
    age: 86400,   // Keep failed jobs for 24 hours
  },
}

Adding Jobs

import { addJob, QUEUE_NAMES } from '@/lib/queue/bullmq-queue';

// Add job to queue
const job = await addJob(
  QUEUE_NAMES.EMAIL_DELIVERY,
  'send-welcome-email',
  {
    userId: 'user_123',
    email: 'user@example.com',
    templateId: 'welcome',
  },
  {
    priority: 1,        // Higher priority processed first
    delay: 5000,        // Delay by 5 seconds
    attempts: 5,        // Override default attempts
    jobId: 'unique-id', // Idempotency key
  }
);

Creating Workers

import { createWorker, QUEUE_NAMES } from '@/lib/queue/bullmq-queue';
import type { Job } from 'bullmq';

interface EmailJobData {
  userId: string;
  email: string;
  templateId: string;
}

// Create worker with processor
const worker = createWorker<EmailJobData, { success: boolean }>(
  QUEUE_NAMES.EMAIL_DELIVERY,
  async (job: Job<EmailJobData>) => {
    const { userId, email, templateId } = job.data;

    // Process the job
    await emailService.send(email, templateId);

    return { success: true };
  },
  {
    concurrency: 5,  // Process 5 jobs concurrently
    limiter: {
      max: 100,      // Max 100 jobs
      duration: 1000, // Per 1 second (rate limiting)
    },
  }
);

// Worker events
worker.on('completed', (job) => {
  logger.info('Job completed', { jobId: job.id });
});

worker.on('failed', (job, err) => {
  logger.error('Job failed', { jobId: job?.id, error: err.message });
});

Graceful Fallback

BullMQ gracefully falls back when Redis is unavailable:

import { isBullMQAvailable, getFallbackQueueStatus } from '@/lib/queue/bullmq-queue';

if (!isBullMQAvailable()) {
  // Falls back to database-backed DeferredTask table
  const status = await getFallbackQueueStatus();
}

Metrics

import { getQueueMetrics, getAllQueueMetrics } from '@/lib/queue/bullmq-queue';

// Get metrics for specific queue
const metrics = await getQueueMetrics(QUEUE_NAMES.EMAIL_DELIVERY);
console.log(metrics);
// {
//   name: 'email-delivery',
//   waiting: 45,
//   active: 5,
//   completed: 1234,
//   failed: 12,
//   delayed: 3,
//   paused: 0,
//   total: 48,
//   status: 'healthy',
//   failureRate: 0.010,
//   throughputPerMinute: 20,
//   oldestJobAge: 1234
// }

// Get all queue metrics
const allMetrics = await getAllQueueMetrics();

---

Database Job Pattern

Overview

Database-backed job queues provide durability and simplicity without external dependencies. Jobs are stored in PostgreSQL tables and processed via polling.

Use cases: Review syncing, call log ingestion, any task requiring guaranteed processing.

Architecture

┌─────────────────────────────────────────────────┐
│            Database Job Pattern                  │
├─────────────────────────────────────────────────┤
│                                                   │
│  1. Job Enqueuing                                │
│     └─> Insert job with status='pending'         │
│                                                   │
│  2. Job Claiming (Compare-and-Set)               │
│     └─> UPDATE ... WHERE status='pending'        │
│         RETURNING * (atomic claim)               │
│                                                   │
│  3. Job Processing                               │
│     └─> Execute business logic                   │
│                                                   │
│  4. Job Completion                               │
│     ├─> Success: status='completed'              │
│     └─> Failure: Retry with backoff              │
│                                                   │
│  5. Dead Letter Queue                            │
│     └─> After max attempts: status='dead_letter' │
│                                                   │
└─────────────────────────────────────────────────┘

Example: Google Review Sync Jobs

Service: /lib/jobs/googleReviewSyncJobService.ts Table: GoogleReviewSyncJob Cron: /app/api/cron/google-reviews/route.ts

Job Statuses

type JobStatus = 'pending' | 'running' | 'completed' | 'failed' | 'dead_letter';

Job Configuration

const JOB_CLAIM_OPTIONS = {
  maxAttempts: 5,
  baseBackoffMs: 60_000,           // 1 minute base delay
  maxJobsPerClaim: 1,
  maxRetryDelayMs: 12 * 60 * 60 * 1000, // 12 hours max
};

Enqueue Jobs

import { googleReviewSyncJobService } from '@/lib/jobs/googleReviewSyncJobService';

// Enqueue single job
const job = await googleReviewSyncJobService.enqueueJob(
  googleConnectionId,
  {
    scheduledAt: new Date(Date.now() + 60000), // Schedule 1 min from now
    priority: 1,
    payload: { fullSync: true },
  }
);

// Bulk enqueue
const connectionIds = ['conn_1', 'conn_2', 'conn_3'];
const count = await googleReviewSyncJobService.enqueueJobs(connectionIds);

Claim and Process Jobs

// Claim next job (atomic compare-and-set)
const job = await googleReviewSyncJobService.claimNextJob();

if (job) {
  try {
    // Process job
    const result = await googleReviewService.syncReviews(job.googleConnectionId);

    // Mark complete
    await googleReviewSyncJobService.completeJob(job.id, result);
  } catch (error) {
    // Handle failure (auto-retry with backoff)
    await googleReviewSyncJobService.failJob(job.id, error);
  }
}

Stuck Job Recovery

// Re-queue jobs that have been running too long (15 min default)
const requeuedCount = await googleReviewSyncJobService.requeueStuckJobs(15);

Queue Metrics

const metrics = await googleReviewSyncJobService.getQueueMetrics();
// {
//   pending: 23,
//   running: 2,
//   failed: 1,
//   deadLetter: 0
// }

Example: Call Log Sync Jobs

Service: /lib/jobs/callLogSyncJobService.ts Tables: CallLogSyncState, CallLogSyncJob Cron: /app/api/cron/call-logs/route.ts

This pattern uses a state table to track sync cursors and a job table for processing.

Create Sync State

import { callLogSyncJobService } from '@/lib/jobs/callLogSyncJobService';

// Ensure state exists for client/provider/phone
const state = await callLogSyncJobService.ensureState({
  clientId: 'client_123',
  provider: 'twilio',
  phoneNumber: '+14155551234',
  connectionIntegrationId: 'conn_int_456',
});

Enqueue Job for State

// Enqueue job for specific state
const jobId = await callLogSyncJobService.enqueueState(state.id);

// Enqueue jobs for all active states
const enqueuedCount = await callLogSyncJobService.enqueueAllActiveStates();

Update Sync Cursor

// After successful sync, update cursor
await callLogSyncJobService.updateStateCursor(
  stateId,
  { lastCallId: 'CA1234567890', timestamp: '2025-12-13T10:00:00Z' },
  new Date()
);

Pause/Resume States

// Pause sync state
await callLogSyncJobService.pauseState(stateId);

// Resume sync state
await callLogSyncJobService.resumeState(stateId);

---

Cron Jobs

Overview

Vercel Cron jobs run scheduled tasks at defined intervals. All cron endpoints are in /app/api/cron/.

Configuration: vercel.json

Cron Schedule

{
  "crons": [
    {
      "path": "/api/health/edge",
      "schedule": "*/15 * * * *"  // Every 15 minutes
    },
    {
      "path": "/api/cron/google-reviews",
      "schedule": "*/15 * * * *"  // Every 15 minutes
    },
    {
      "path": "/api/cron/message-status",
      "schedule": "*/5 * * * *"   // Every 5 minutes
    },
    {
      "path": "/api/cron/orphan-cleanup",
      "schedule": "0 */6 * * *"   // Every 6 hours
    },
    {
      "path": "/api/cron/dunning",
      "schedule": "0 8 * * *"     // Daily at 8 AM
    },
    {
      "path": "/api/cron/lead-settlement",
      "schedule": "0 6 * * *"     // Daily at 6 AM
    },
    {
      "path": "/api/cron/presence-cleanup",
      "schedule": "*/5 * * * *"   // Every 5 minutes
    }
  ]
}

Cron Schedule Syntax

┌───────────── minute (0 - 59)
│ ┌───────────── hour (0 - 23)
│ │ ┌───────────── day of month (1 - 31)
│ │ │ ┌───────────── month (1 - 12)
│ │ │ │ ┌───────────── day of week (0 - 6) (Sunday to Saturday)
│ │ │ │ │
│ │ │ │ │
* * * * *

Examples:
*/5 * * * *      → Every 5 minutes
0 */6 * * *      → Every 6 hours
0 8 * * *        → Daily at 8:00 AM
0 0 * * 0        → Weekly on Sunday at midnight
0 0 1 * *        → Monthly on 1st at midnight

Authentication

All cron endpoints require authentication via CRON_SECRET:

// Timing-safe comparison to prevent timing attacks
function verifyCronSecret(request: Request): boolean {
  const authHeader = request.headers.get('authorization');
  const cronSecret = process.env['CRON_SECRET'];

  if (!cronSecret) {
    return process.env['NODE_ENV'] !== 'production';
  }

  return authHeader === `Bearer ${cronSecret}`;
}

export async function GET(request: Request) {
  if (!verifyCronSecret(request)) {
    return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
  }

  // Process cron job...
}

Cron Job Pattern

export async function GET(request: Request) {
  // 1. Verify authentication
  if (!verifyCronSecret(request)) {
    return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
  }

  try {
    logger.info('Starting cron job');
    const startTime = Date.now();

    // 2. Re-queue stuck jobs
    const stuckCount = await jobService.requeueStuckJobs();

    // 3. Enqueue new jobs
    const enqueuedCount = await jobService.enqueueAllActiveStates();

    // 4. Process batch of jobs
    const MAX_JOBS_PER_RUN = 10;
    const processed = [];

    for (let i = 0; i < MAX_JOBS_PER_RUN; i++) {
      const job = await jobService.claimNextJob();
      if (!job) break;

      try {
        const result = await processJob(job);
        await jobService.completeJob(job.id, result);
        processed.push({ jobId: job.id, status: 'completed', result });
      } catch (error) {
        await jobService.failJob(job.id, error);
        processed.push({ jobId: job.id, status: 'failed', error: error.message });
      }
    }

    // 5. Get metrics
    const metrics = await jobService.getQueueMetrics();

    // 6. Alert on anomalies
    if (metrics.deadLetter > 0) {
      logger.error('Jobs in dead letter queue', metrics);
    }
    if (metrics.pending > ALERT_THRESHOLD) {
      logger.warn('Queue backlog growing', metrics);
    }

    const duration = Date.now() - startTime;
    logger.info('Cron job completed', { duration, enqueuedCount, processedCount: processed.length });

    return NextResponse.json({
      success: true,
      duration,
      stuckRequeued: stuckCount,
      jobsEnqueued: enqueuedCount,
      jobsProcessed: processed.length,
      processed,
      metrics,
    });
  } catch (error) {
    logger.error('Cron job failed', { error });
    return NextResponse.json(
      { success: false, error: error.message },
      { status: 500 }
    );
  }
}

Sentry Monitoring

Use Sentry.withMonitor for cron health tracking:

import * as Sentry from '@sentry/nextjs';

export async function GET(request: NextRequest) {
  // Verify auth...

  return Sentry.withMonitor(
    'data-retention-daily',
    async () => {
      // Execute cron logic...
      return NextResponse.json({ success: true });
    },
    {
      schedule: { type: 'crontab', value: '0 3 * * *' },
      checkinMargin: 5,      // Minutes of leeway for execution
      maxRuntime: 30,        // Max minutes before timeout alert
      timezone: 'America/Los_Angeles',
    }
  );
}

---

Retry Strategies

Exponential Backoff

All job patterns use exponential backoff to prevent overwhelming systems during failures.

BullMQ Retry

// Automatic exponential backoff
{
  attempts: 3,
  backoff: {
    type: 'exponential',
    delay: 1000, // 1 second base
  }
}

// Retry schedule:
// Attempt 1: Immediate
// Attempt 2: 1 second delay
// Attempt 3: 2 seconds delay
// Attempt 4: 4 seconds delay

Database Job Retry

const JOB_CLAIM_OPTIONS = {
  maxAttempts: 5,
  baseBackoffMs: 60_000,           // 1 minute
  maxRetryDelayMs: 12 * 60 * 60 * 1000, // 12 hours
};

function getBackoffDelay(attemptCount: number): number {
  const exponent = Math.max(0, attemptCount);
  const delay = baseBackoffMs * Math.pow(2, exponent);
  return Math.min(delay, maxRetryDelayMs);
}

// Retry schedule:
// Attempt 1: Immediate
// Attempt 2: 1 minute delay
// Attempt 3: 2 minutes delay
// Attempt 4: 4 minutes delay
// Attempt 5: 8 minutes delay
// Attempt 6+: 12 hours delay (capped)

Custom Retry Logic

async function failJob(jobId: string, error: unknown): Promise<void> {
  const job = await prisma.job.findUnique({ where: { id: jobId } });
  if (!job) return;

  const delay = this.getBackoffDelay(job.attemptCount);
  const hasAttemptsRemaining = job.attemptCount < maxAttempts;

  const nextStatus = hasAttemptsRemaining ? 'pending' : 'dead_letter';

  await prisma.job.update({
    where: { id: jobId },
    data: {
      status: nextStatus,
      scheduledAt: hasAttemptsRemaining
        ? new Date(Date.now() + delay)
        : job.scheduledAt,
      lastErrorAt: new Date(),
      lastErrorCode: error instanceof Error && 'code' in error
        ? String(error.code)
        : undefined,
      lastErrorMessage: error instanceof Error
        ? error.message
        : String(error),
    },
  });
}

Dead Letter Queue (DLQ)

Jobs that exceed max retry attempts are moved to a dead letter queue for manual review.

// Check for dead letter jobs
const metrics = await jobService.getQueueMetrics();

if (metrics.deadLetter > 0) {
  logger.error('Jobs in dead letter queue require manual intervention', {
    deadLetterCount: metrics.deadLetter,
  });

  // Alert ops team
  await alertService.sendSlackAlert({
    channel: '#ops-alerts',
    message: `${metrics.deadLetter} jobs in dead letter queue`,
  });
}

Rate Limiting

Prevent overwhelming external APIs:

// BullMQ rate limiter
createWorker(
  QUEUE_NAMES.EMAIL_DELIVERY,
  processor,
  {
    limiter: {
      max: 100,      // Max 100 jobs
      duration: 1000, // Per 1 second
    },
  }
);

// Custom rate limiting
const RATE_LIMIT = {
  maxJobsPerRun: 10,
  delayBetweenJobs: 100, // ms
};

for (let i = 0; i < RATE_LIMIT.maxJobsPerRun; i++) {
  const job = await claimNextJob();
  if (!job) break;

  await processJob(job);
  await sleep(RATE_LIMIT.delayBetweenJobs);
}

---

Error Handling

Error Classification

interface JobError {
  code: string;
  message: string;
  severity: 'transient' | 'permanent';
  retryable: boolean;
}

function classifyError(error: unknown): JobError {
  if (error instanceof NetworkError) {
    return {
      code: 'NETWORK_ERROR',
      message: error.message,
      severity: 'transient',
      retryable: true,
    };
  }

  if (error instanceof ValidationError) {
    return {
      code: 'VALIDATION_ERROR',
      message: error.message,
      severity: 'permanent',
      retryable: false,
    };
  }

  // Default: treat as transient
  return {
    code: 'UNKNOWN_ERROR',
    message: String(error),
    severity: 'transient',
    retryable: true,
  };
}

Structured Error Logging

async function failJob(jobId: string, error: unknown): Promise<void> {
  const classified = classifyError(error);

  logger.error('Job failed', {
    jobId,
    errorCode: classified.code,
    errorMessage: classified.message,
    severity: classified.severity,
    retryable: classified.retryable,
    attemptCount: job.attemptCount,
    nextStatus: classified.retryable ? 'pending' : 'dead_letter',
  });

  // Store structured error in database
  await prisma.job.update({
    where: { id: jobId },
    data: {
      lastErrorCode: classified.code,
      lastErrorMessage: classified.message,
      lastErrorAt: new Date(),
      errorMetadata: {
        severity: classified.severity,
        retryable: classified.retryable,
        attemptCount: job.attemptCount,
      } as Prisma.JsonObject,
    },
  });
}

Circuit Breaker Pattern

class CircuitBreaker {
  private failures = 0;
  private lastFailureTime: Date | null = null;
  private state: 'closed' | 'open' | 'half-open' = 'closed';

  constructor(
    private threshold = 5,
    private timeout = 60000 // 1 minute
  ) {}

  async execute<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === 'open') {
      if (Date.now() - this.lastFailureTime!.getTime() > this.timeout) {
        this.state = 'half-open';
      } else {
        throw new Error('Circuit breaker is open');
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  private onSuccess(): void {
    this.failures = 0;
    this.state = 'closed';
  }

  private onFailure(): void {
    this.failures++;
    this.lastFailureTime = new Date();

    if (this.failures >= this.threshold) {
      this.state = 'open';
      logger.error('Circuit breaker opened', {
        failures: this.failures,
        threshold: this.threshold,
      });
    }
  }
}

// Usage
const breaker = new CircuitBreaker();

await breaker.execute(async () => {
  return await externalApiCall();
});

---

Monitoring and Debugging

Prometheus Metrics

All background jobs emit Prometheus metrics for monitoring.

Metrics Exposed:

petunia_queue_jobs_total{queue, status}
petunia_queue_job_duration_seconds{queue}
petunia_queue_jobs_active{queue}
petunia_queue_jobs_waiting{queue}
petunia_queue_jobs_failed{queue}

Recording Metrics:

import { recordQueueJob, updateQueueCounts } from '@/lib/services/monitoring/prometheusMetrics';

// Record job completion
recordQueueJob('email-delivery', 'completed', durationSeconds);

// Update queue counts
updateQueueCounts('email-delivery', active, waiting, failed);

Logging Best Practices

// Start of job
logger.info('Starting job processing', {
  jobId: job.id,
  jobType: job.type,
  attemptCount: job.attemptCount,
  scheduledAt: job.scheduledAt,
});

// During processing
logger.debug('Processing step completed', {
  jobId: job.id,
  step: 'fetch-data',
  recordsProcessed: 100,
});

// On success
logger.info('Job completed successfully', {
  jobId: job.id,
  durationMs: Date.now() - startTime,
  result: {
    recordsProcessed: 100,
    recordsCreated: 50,
    recordsUpdated: 30,
    recordsSkipped: 20,
  },
});

// On failure
logger.error('Job failed', {
  jobId: job.id,
  error: error.message,
  errorCode: error.code,
  attemptCount: job.attemptCount,
  willRetry: job.attemptCount < maxAttempts,
  nextRetryAt: new Date(Date.now() + backoffDelay),
});

Debugging Stuck Jobs

-- Find jobs stuck in 'running' state
SELECT
  id,
  "callLogSyncStateId",
  status,
  "attemptCount",
  "startedAt",
  "scheduledAt",
  NOW() - "startedAt" AS running_duration
FROM "CallLogSyncJob"
WHERE status = 'running'
  AND "startedAt" < NOW() - INTERVAL '15 minutes'
ORDER BY "startedAt" ASC;

-- Find jobs in dead letter queue
SELECT
  id,
  "callLogSyncStateId",
  "attemptCount",
  "lastErrorCode",
  "lastErrorMessage",
  "lastErrorAt"
FROM "CallLogSyncJob"
WHERE status = 'dead_letter'
ORDER BY "lastErrorAt" DESC;

-- Queue backlog analysis
SELECT
  status,
  COUNT(*) as count,
  MIN("scheduledAt") as oldest_job,
  MAX("scheduledAt") as newest_job
FROM "CallLogSyncJob"
GROUP BY status;

Job Metrics Queries

// Get queue health metrics
const metrics = await jobService.getQueueMetrics();

console.log(`
Queue Health Report:
  Pending:      ${metrics.pending}
  Running:      ${metrics.running}
  Completed:    ${metrics.completed}
  Failed:       ${metrics.failed}
  Dead Letter:  ${metrics.deadLetter}

  Oldest Pending: ${metrics.oldestPending}
  Active States:  ${metrics.activeStates}
  Error States:   ${metrics.errorStates}
`);

// Alert thresholds
const ALERTS = {
  pendingBacklog: 25,
  deadLetterThreshold: 1,
  errorStateThreshold: 5,
};

if (metrics.pending > ALERTS.pendingBacklog) {
  logger.warn('Queue backlog growing', { pending: metrics.pending });
}

if (metrics.deadLetter > ALERTS.deadLetterThreshold) {
  logger.error('Jobs in dead letter queue', { count: metrics.deadLetter });
}

Sentry Error Tracking

import * as Sentry from '@sentry/nextjs';

try {
  await processJob(job);
} catch (error) {
  Sentry.captureException(error, {
    tags: {
      jobType: 'google-review-sync',
      jobId: job.id,
      attemptCount: job.attemptCount,
    },
    contexts: {
      job: {
        id: job.id,
        googleConnectionId: job.googleConnectionId,
        scheduledAt: job.scheduledAt,
        attemptCount: job.attemptCount,
      },
    },
  });

  throw error;
}

---

Best Practices

1. Idempotency

Jobs should be idempotent - running the same job multiple times should produce the same result.

// Use unique job IDs to prevent duplicates
await addJob(
  QUEUE_NAMES.EMAIL_DELIVERY,
  'send-welcome-email',
  { userId },
  {
    jobId: `welcome-email-${userId}`, // Prevents duplicate emails
  }
);

// Check if job already processed
const existingJob = await prisma.job.findUnique({
  where: {
    userId_taskType: {
      userId,
      taskType: 'welcome-email'
    }
  },
});

if (existingJob?.completedAt) {
  logger.info('Job already completed, skipping', { jobId: existingJob.id });
  return;
}

2. Small, Focused Jobs

Break large tasks into smaller jobs:

// BAD: Process 10,000 records in one job
await addJob(QUEUE_NAMES.DATA_IMPORT, 'import-all', { recordIds: allRecordIds });

// GOOD: Process in batches
const batchSize = 100;
for (let i = 0; i < allRecordIds.length; i += batchSize) {
  const batch = allRecordIds.slice(i, i + batchSize);
  await addJob(QUEUE_NAMES.DATA_IMPORT, 'import-batch', { recordIds: batch });
}

3. Graceful Degradation

Handle missing dependencies gracefully:

// BullMQ not available? Fall back to database jobs
if (!isBullMQAvailable()) {
  logger.warn('BullMQ unavailable, using database job fallback');
  await databaseJobService.enqueueJob(jobData);
  return;
}

await addJob(QUEUE_NAMES.NOTIFICATION, 'send-notification', jobData);

4. Job Timeouts

Set reasonable timeouts to prevent zombie jobs:

const MAX_JOB_RUNTIME_MINUTES = 15;

async function processJob(job: Job): Promise<void> {
  const timeout = setTimeout(() => {
    throw new Error(`Job exceeded ${MAX_JOB_RUNTIME_MINUTES} minute timeout`);
  }, MAX_JOB_RUNTIME_MINUTES * 60 * 1000);

  try {
    await actualProcessing(job);
  } finally {
    clearTimeout(timeout);
  }
}

5. Monitoring and Alerting

Set up alerts for job anomalies:

const ALERT_THRESHOLDS = {
  pendingBacklog: 25,
  deadLetterJobs: 1,
  failureRate: 0.1, // 10%
};

const metrics = await getQueueMetrics();

if (metrics.failed / (metrics.completed + metrics.failed) > ALERT_THRESHOLDS.failureRate) {
  await sendAlert({
    severity: 'warning',
    message: `Queue ${metrics.name} failure rate above ${ALERT_THRESHOLDS.failureRate * 100}%`,
  });
}

6. Resource Cleanup

Always clean up resources after job processing:

async function processJob(job: Job): Promise<void> {
  const tempFiles: string[] = [];

  try {
    const tempFile = await downloadFile(job.data.url);
    tempFiles.push(tempFile);

    await processFile(tempFile);
  } finally {
    // Clean up temp files
    for (const file of tempFiles) {
      await fs.unlink(file).catch(err => {
        logger.warn('Failed to delete temp file', { file, err });
      });
    }
  }
}

---

Examples

Example 1: Email Delivery Queue (BullMQ)

// /lib/queue/email-queue.ts
import { addJob, createWorker, QUEUE_NAMES } from '@/lib/queue/bullmq-queue';
import { emailService } from '@/lib/services/email/emailService';

interface EmailJobData {
  to: string;
  subject: string;
  templateId: string;
  templateData: Record<string, any>;
}

// Add email to queue
export async function queueEmail(data: EmailJobData): Promise<void> {
  await addJob(
    QUEUE_NAMES.EMAIL_DELIVERY,
    'send-email',
    data,
    {
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 2000,
      },
      removeOnComplete: true,
    }
  );
}

// Create worker (in separate process/serverless function)
export function startEmailWorker() {
  return createWorker<EmailJobData, { success: boolean }>(
    QUEUE_NAMES.EMAIL_DELIVERY,
    async (job) => {
      const { to, subject, templateId, templateData } = job.data;

      await emailService.send({
        to,
        subject,
        templateId,
        templateData,
      });

      return { success: true };
    },
    {
      concurrency: 5,
      limiter: {
        max: 50,        // Max 50 emails
        duration: 1000, // Per second (respect rate limits)
      },
    }
  );
}

Example 2: Review Sync Cron Job (Database Pattern)

// /app/api/cron/google-reviews/route.ts
import { NextResponse } from 'next/server';
import { googleReviewSyncJobService } from '@/lib/jobs/googleReviewSyncJobService';
import { googleReviewService } from '@/lib/connections/google/googleReviewService';

const MAX_JOBS_PER_RUN = 10;

export async function POST(request: Request) {
  // Verify cron secret
  if (!verifyCronSecret(request)) {
    return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
  }

  // Step 1: Re-queue stuck jobs
  const requeued = await googleReviewSyncJobService.requeueStuckJobs();

  // Step 2: Enqueue jobs for enabled connections
  const connections = await prisma.googleConnection.findMany({
    where: {
      syncEnabled: true,
      ConnectionIntegration: {
        isEnabled: true,
        isConfigured: true,
      },
    },
    select: { id: true },
  });

  const enqueued = await googleReviewSyncJobService.enqueueJobs(
    connections.map(c => c.id)
  );

  // Step 3: Process batch of jobs
  const processed = [];

  for (let i = 0; i < MAX_JOBS_PER_RUN; i++) {
    const job = await googleReviewSyncJobService.claimNextJob();
    if (!job) break;

    try {
      const result = await googleReviewService.syncReviews(job.googleConnectionId);
      await googleReviewSyncJobService.completeJob(job.id, result);
      processed.push({ jobId: job.id, status: 'completed', result });
    } catch (error) {
      await googleReviewSyncJobService.failJob(job.id, error);
      processed.push({
        jobId: job.id,
        status: 'failed',
        error: error.message
      });
    }
  }

  // Step 4: Check metrics and alert
  const metrics = await googleReviewSyncJobService.getQueueMetrics();

  if (metrics.deadLetter > 0) {
    logger.error('Google review queue has dead-letter jobs', metrics);
  }

  return NextResponse.json({
    success: true,
    enqueued,
    requeued,
    processedCount: processed.length,
    processed,
    metrics,
  });
}

Example 3: Scheduled Lead Settlement (Cron)

// /app/api/cron/lead-settlement/route.ts
import { NextResponse } from 'next/server';
import { CreditWalletService } from '@/lib/services/billing/credit-wallet-service';

export async function GET(req: NextRequest) {
  if (!verifyCronSecret(req)) {
    return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
  }

  const now = new Date();
  const dayOfWeek = now.getDay(); // 0 = Sunday, 1 = Monday
  const dayOfMonth = now.getDate();

  // Get wallets with unpaid balances
  const wallets = await prisma.creditWallet.findMany({
    where: {
      unpaidLeadBalanceCents: { gt: 0 },
      leadBillingSchedule: { in: ['daily', 'weekly', 'monthly'] },
    },
  });

  const results = { settled: 0, skipped: 0, failed: 0 };

  for (const wallet of wallets) {
    let shouldSettle = false;

    switch (wallet.leadBillingSchedule) {
      case 'daily':
        shouldSettle = true;
        break;
      case 'weekly':
        shouldSettle = dayOfWeek === 1; // Monday
        break;
      case 'monthly':
        shouldSettle = dayOfMonth === 1; // 1st of month
        break;
    }

    if (!shouldSettle) {
      results.skipped++;
      continue;
    }

    try {
      await CreditWalletService.settleLeadBalance(wallet.clientId);
      results.settled++;
    } catch (error) {
      logger.error('Settlement failed', { clientId: wallet.clientId, error });
      results.failed++;
    }
  }

  return NextResponse.json({ success: true, ...results });
}

---

Summary

Petunia's background job system provides three complementary patterns:

1. BullMQ - Fast, scalable Redis-based queues for high-throughput async tasks
2. Database Jobs - Durable, poll-based pattern for reliable task processing
3. Cron Jobs - Scheduled maintenance tasks via Vercel Cron

All patterns include:

• Exponential backoff retry strategies
• Dead letter queues for failed jobs
• Prometheus metrics for monitoring
• Sentry integration for error tracking
• Graceful degradation and fallbacks

Key Principles:

• Idempotent job design
• Small, focused jobs
• Comprehensive logging
• Proactive monitoring and alerting
• Resource cleanup

For questions or issues, see:

• ARCHITECTURE.md
• MONITORING.md (if exists)
• Slack: #engineering-jobs