Building a Content Queue System with Redis + Node.js: What They Don't Tell You

// The naive approach. Works at 5 req/min. Collapses at 50.
for (const article of articles) {
  const content = await generateContent(article.prompt);
  await db.save({ id: article.id, content });
}

npm install bullmq ioredis

// lib/redis.js
import { Redis } from "ioredis";

// BullMQ requires TWO separate Redis connections:
// one for the queue (subscriber) and one for the worker (publisher).
// Reusing a single connection will cause silent failures under load.
export function createRedisConnection() {
  return new Redis({
    host: process.env.REDIS_HOST || "localhost",
    port: parseInt(process.env.REDIS_PORT || "6379"),
    maxRetriesPerRequest: null, // Required by BullMQ — do not omit
    enableReadyCheck: false,    // Prevents startup race conditions
    lazyConnect: true,          // Don't connect until first command
  });
}

// lib/contentQueue.js
import { Queue } from "bullmq";
import { createRedisConnection } from "./redis.js";

export const contentQueue = new Queue("content-generation", {
  connection: createRedisConnection(),
  defaultJobOptions: {
    attempts: 3,           // Retry up to 3 times before marking failed
    backoff: {
      type: "exponential",
      delay: 2000,         // First retry: 2s. Second: 4s. Third: 8s.
    },
    removeOnComplete: {
      age: 86400,          // Keep completed jobs for 24h for debugging
      count: 1000,         // But cap at 1000 to prevent memory bloat
    },
    removeOnFail: false,   // Never auto-remove failed jobs — you want them
  },
});

// api/generateArticles.js  
import { contentQueue } from "../lib/contentQueue.js";

export async function queueArticleBatch(articles) {
  const jobs = articles.map((article) => ({
    name: "generate-article",
    data: {
      articleId: article.id,
      keyword: article.keyword,
      wordCount: article.wordCount || 1000,
      userId: article.userId,
    },
    opts: {
      // Job ID deduplication: same article won't queue twice
      // even if the endpoint is called twice (e.g., double-submit)
      jobId: `article-${article.id}`,
      priority: article.isPriority ? 1 : 10, // Lower number = higher priority
    },
  }));

  // addBulk is significantly faster than calling add() in a loop.
  // Internally it uses a single Redis pipeline instead of N round trips.
  const result = await contentQueue.addBulk(jobs);

  return result.map((job) => ({ jobId: job.id, articleId: job.data.articleId }));
}

// workers/contentWorker.js
import { Worker } from "bullmq";
import { createRedisConnection } from "../lib/redis.js";
import { generateWithRetry } from "../lib/openai.js";
import { db } from "../lib/db.js";

const worker = new Worker(
  "content-generation",
  async (job) => {
    const { articleId, keyword, wordCount, userId } = job.data;

    // Update progress so the UI can show a spinner instead of nothing
    await job.updateProgress(10);

    const content = await generateWithRetry({ keyword, wordCount });

    await job.updateProgress(80);

    await db.articles.update({
      where: { id: articleId },
      data: {
        content,
        status: "completed",
        completedAt: new Date(),
      },
    });

    await job.updateProgress(100);

    return { articleId, wordCount: content.split(" ").length };
  },
  {
    connection: createRedisConnection(), // Separate connection from the queue
    concurrency: 5, // Process 5 jobs simultaneously
    limiter: {
      max: 10,         // Max 10 jobs processed
      duration: 1000,  // per 1000ms (per second)
    },
  }
);

// Worker event handlers — non-negotiable for production
worker.on("completed", (job, result) => {
  console.log(`[${job.id}] Completed: ${result.wordCount} words for article ${result.articleId}`);
});

worker.on("failed", (job, error) => {
  console.error(`[${job?.id}] Failed after ${job?.attemptsMade} attempts:`, error.message);
  // Add your alerting here — Slack, PagerDuty, whatever you use
});

worker.on("error", (error) => {
  // This catches connection errors, not job errors
  console.error("Worker connection error:", error);
});

// lib/openai.js
import OpenAI from "openai";

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

export async function generateWithRetry({ keyword, wordCount }, retries = 3) {
  for (let attempt = 1; attempt <= retries; attempt++) {
    try {
      const response = await openai.chat.completions.create({
        model: "gpt-4o",
        messages: [
          {
            role: "system",
            content: "You are a professional content writer. Respond with article content only, no preamble.",
          },
          {
            role: "user",
            content: `Write a ${wordCount}-word article about "${keyword}".
Include: engaging introduction, 3 H2 sections, practical examples, conclusion.
Tone: professional but approachable.`,
          },
        ],
        max_tokens: Math.ceil(wordCount * 1.5), // ~1.5 tokens per word average
        temperature: 0.7,
      });

      return response.choices[0].message.content;

    } catch (error) {
      const isRateLimit = error.status === 429;
      const isServerError = error.status >= 500;
      const isRetryable = isRateLimit || isServerError;

      if (!isRetryable || attempt === retries) {
        throw error; // Surface to BullMQ for job-level retry
      }

      // Exponential backoff: 1s, 2s, 4s
      const delay = Math.pow(2, attempt - 1) * 1000;
      console.warn(`OpenAI ${error.status} on attempt ${attempt}. Retrying in ${delay}ms...`);
      await new Promise((resolve) => setTimeout(resolve, delay));
    }
  }
}

// api/admin/jobs.js
import { contentQueue } from "../../lib/contentQueue.js";

// GET /admin/jobs/failed
export async function getFailedJobs(req, res) {
  const failedJobs = await contentQueue.getFailed(0, 50); // First 50 failed jobs

  res.json({
    count: failedJobs.length,
    jobs: failedJobs.map((job) => ({
      id: job.id,
      articleId: job.data.articleId,
      failedReason: job.failedReason,
      attemptsMade: job.attemptsMade,
      timestamp: new Date(job.timestamp).toISOString(),
    })),
  });
}

// POST /admin/jobs/:jobId/retry
export async function retryFailedJob(req, res) {
  const { jobId } = req.params;
  const job = await contentQueue.getJob(jobId);

  if (!job) return res.status(404).json({ error: "Job not found" });

  await job.retry();
  res.json({ message: `Job ${jobId} requeued`, status: "waiting" });
}

// POST /admin/jobs/retry-all-failed
export async function retryAllFailed(req, res) {
  const failedJobs = await contentQueue.getFailed(0, 100);
  await Promise.all(failedJobs.map((job) => job.retry()));

  res.json({ message: `${failedJobs.length} jobs requeued` });
}

const worker = new Worker("content-generation", processor, {
  connection: createRedisConnection(),
  lockDuration: 60000,    // 60 seconds
  lockRenewTime: 20000,   // Renew every 20s (should be lockDuration / 3)
});

// api/health/queue.js
export async function getQueueHealth(req, res) {
  const [waiting, active, completed, failed, delayed] = await Promise.all([
    contentQueue.getWaitingCount(),
    contentQueue.getActiveCount(),
    contentQueue.getCompletedCount(),
    contentQueue.getFailedCount(),
    contentQueue.getDelayedCount(),
  ]);

  const isHealthy = failed === 0 || (failed / (completed + failed)) < 0.05; // < 5% failure rate

  res.status(isHealthy ? 200 : 503).json({
    status: isHealthy ? "healthy" : "degraded",
    queue: { waiting, active, completed, failed, delayed },
    failureRate: completed + failed > 0
      ? `${((failed / (completed + failed)) * 100).toFixed(1)}%`
      : "0%",
  });
}