Job Queue

// 1) Naive (UNSAFE) queue — works until a worker crashes mid-job
import Redis from 'ioredis';
const redis = new Redis(process.env.REDIS_URL);

// Producer
await redis.lpush('jobs', JSON.stringify({ type: 'email', to: 'a@b.com' }));

// Worker
async function worker() {
    while (true) {
        const raw = await redis.brpop('jobs', 0);     // blocks until a job appears
        const job = JSON.parse(raw[1]);
        await handle(job);                              // ❌ if process dies here, job lost
    }
}

// 2) Reliable queue — BRPOPLPUSH (or BLMOVE) to a processing list
async function reliableWorker(workerId) {
    const processingKey = `processing:${workerId}`;
    while (true) {
        const raw = await redis.brpoplpush('jobs', processingKey, 0);
        try {
            await handle(JSON.parse(raw));
            await redis.lrem(processingKey, 1, raw);    // ack — remove from processing
        } catch (e) {
            // leave it on processingKey; cleanup script moves it to retry/dead-letter
            log.error(e, 'job failed');
        }
    }
}

// On startup, requeue any leftover jobs from this worker's processing list:
await redis.eval(`
while true do
    local v = redis.call('RPOPLPUSH', KEYS[1], KEYS[2])
    if not v then break end
end
return 1
`, 2, processingKey, 'jobs');

// 3) Janitor — reclaim stuck jobs from dead workers
// Maintain a heartbeat key per worker; periodically check + recover.

setInterval(async () => {
    await redis.set(`hb:${workerId}`, Date.now(), 'EX', 60);
}, 15_000);

async function reclaimStuck() {
    const workers = await redis.keys('hb:*');
    const known = new Set(workers.map((k) => k.slice(3)));
    const processings = await redis.keys('processing:*');
    for (const pk of processings) {
        const id = pk.slice('processing:'.length);
        if (!known.has(id)) {
            // worker crashed → move all jobs back
            await redis.eval(`
while true do
    local v = redis.call('RPOPLPUSH', KEYS[1], KEYS[2])
    if not v then break end
end
redis.call('DEL', KEYS[1])
return 1
`, 2, pk, 'jobs');
        }
    }
}

// Run reclaimStuck() every minute in a janitor process.

// 4) Retries + dead-letter
async function workerWithRetries(workerId) {
    while (true) {
        const raw = await redis.brpoplpush('jobs', `processing:${workerId}`, 0);
        const job = JSON.parse(raw);
        try {
            await handle(job);
            await redis.lrem(`processing:${workerId}`, 1, raw);
        } catch (e) {
            job.attempts = (job.attempts ?? 0) + 1;
            if (job.attempts >= 5) {
                await redis.lpush('jobs:deadletter', JSON.stringify(job));
            } else {
                await redis.lpush('jobs:retry', JSON.stringify(job));
            }
            await redis.lrem(`processing:${workerId}`, 1, raw);
        }
    }
}

// A scheduler periodically moves jobs from 'jobs:retry' back to 'jobs' with delay.

// 5) Delayed jobs — sorted set as priority queue
const score = Date.now() + 60_000;                  // run in 1 minute
await redis.zadd('jobs:delayed', score, JSON.stringify(job));

// Mover process
setInterval(async () => {
    const now = Date.now();
    const ready = await redis.zrangebyscore('jobs:delayed', 0, now);
    if (ready.length === 0) return;
    const multi = redis.multi();
    multi.zrem('jobs:delayed', ...ready);
    multi.lpush('jobs', ...ready);
    await multi.exec();
}, 1000);

// 6) Redis Streams (modern; 5.0+)
// Streams add consumer groups, ack semantics, message IDs, persistence.
await redis.xadd('jobs', '*', 'type', 'email', 'to', 'a@b.com');

await redis.xgroupCreate('jobs', 'workers', '$', 'MKSTREAM');
const entries = await redis.xreadgroup('GROUP', 'workers', 'worker-1',
    'COUNT', 10, 'BLOCK', 5000, 'STREAMS', 'jobs', '>');
for (const [, msgs] of entries ?? []) {
    for (const [id, fields] of msgs) {
        await handle(toObject(fields));
        await redis.xack('jobs', 'workers', id);
    }
}

// XPENDING + XCLAIM lets you recover messages stuck with crashed workers — built-in equivalent of the reliable pattern.

// Streams support PEL (pending entries list), trim by MAXLEN, message retention, consumer scaling.

// 7) Production libraries
// Most teams DON'T roll their own. Use a battle-tested library:
//   • BullMQ (Node)        — modern, TS-friendly, dashboard, jobs.add/process API
//   • Sidekiq (Ruby)       — gold standard for Ruby; uses Redis
//   • Resque (Ruby)         — older sibling
//   • RQ (Python)            — simple
//   • Celery + Redis broker  — Python; heavy
//   • Asynq (Go)             — Sidekiq-style for Go
//
// They handle reliability, retries, scheduling, rate limiting, priority, dashboards out of the box.

// 8) BullMQ example
import { Queue, Worker } from 'bullmq';
const connection = { host: 'localhost', port: 6379 };
const emailQueue = new Queue('email', { connection });

await emailQueue.add('welcome', { to: 'mara@example.com' }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 } });

const worker = new Worker('email', async (job) => {
    await sendEmail(job.data.to);
}, { connection, concurrency: 10 });

worker.on('failed', (job, err) => console.error(job.id, err));
worker.on('completed', (job) => console.log(job.id, 'done'));

// 9) Anti-patterns
// • Naive LPUSH + BRPOP with auto-ack → lost jobs on crash
// • Worker stores job in local memory before ack → same problem
// • Single worker without max-attempts → infinite retry loop on poisoned message
// • No dead-letter queue → can't inspect what failed
// • Sharing one Redis between queue + cache + sessions → eviction policy mismatch
// • Treating Redis as 'durable enough' — AOF + replicas yes; treat as ephemeral for finance-grade exactly-once

// 10) When NOT to use Redis as a queue
//   • Need exactly-once semantics + complex routing → reach for RabbitMQ / Kafka
//   • Multi-tenant SaaS with strict isolation → dedicated queue per tenant or Kafka partitions
//   • Workflows + scatter/gather → Temporal / AWS Step Functions / Argo
//
// Redis queues are excellent for at-least-once jobs at moderate scale.

// 11) Monitoring
// • LLEN jobs                                — backlog
// • LLEN processing:* and stuck count        — workers behind
// • LLEN jobs:deadletter                      — failed jobs to investigate
// • XINFO STREAM jobs / XPENDING for streams
// • App-level counters: jobs_processed_total, jobs_failed_total, retry_total

// 12) Common bugs
// • Forgetting LREM after success → processing list grows; memory leak
// • Worker keeps two Redis connections active — uses subscription + commands; needs separate clients
// • Lua script timeout → whole Redis blocked
// • Streams without trimming → unbounded growth; use MAXLEN ~ approximate
// • RDB-only persistence with bursty writes → can lose recent jobs on crash; enable AOF
// • Long-running jobs without heartbeat → janitor reclaims while still in progress; periodic ping or chunking
// • Race between janitor and worker → use SHA-locked Lua to make reclaim atomic

# Producer
RPUSH jobs $jsonPayload
# Worker
BLPOP jobs 0           # blocks until a job arrives