Job Queue Architecture

export function getChatQueue(): Queue<ChatJobData, ChatJobResult> {
  return new Queue<ChatJobData, ChatJobResult>("chat", {
    connection: getBullMQConnection(),
    defaultJobOptions: {
      // Retry configuration
      attempts: 3,
      backoff: {
        type: "exponential",
        delay: 1000, // 1s → 2s → 4s
      },
      // Job cleanup
      removeOnComplete: {
        age: 3600, // Keep for 1 hour
        count: 1000,
      },
      removeOnFail: {
        age: 86400, // Keep failed for 24 hours
      },
    },
  });
}

export function getDeepResearchQueue(): Queue<DeepResearchJobData, DeepResearchJobResult> {
  return new Queue<DeepResearchJobData, DeepResearchJobResult>("deep-research", {
    connection: getBullMQConnection(),
    defaultJobOptions: {
      // Fewer retries for long jobs
      attempts: 2,
      backoff: {
        type: "exponential",
        delay: 5000, // 5s → 10s
      },
      // No timeout - jobs can run 20-30+ minutes
      removeOnComplete: {
        age: 86400, // Keep for 24 hours
        count: 500,
      },
      removeOnFail: {
        age: 604800, // Keep failed for 7 days
      },
    },
  });
}

import Redis from "ioredis";

// BullMQ connection (for queues and workers)
export function getBullMQConnection(): Redis {
  const redisUrl = process.env.REDIS_URL || "redis://localhost:6379";
  
  return new Redis(redisUrl, {
    maxRetriesPerRequest: null, // Required for BullMQ
    retryStrategy: (times) => {
      if (times > 10) return null;
      return Math.min(times * 200, 5000);
    },
    reconnectOnError: (err) => {
      const targetErrors = ["READONLY", "ECONNRESET", "ETIMEDOUT"];
      return targetErrors.some((e) => err.message.includes(e));
    },
  });
}

// Publisher (for notifications)
export function getPublisher(): Redis {
  return new Redis(process.env.REDIS_URL);
}

// Subscriber (for notifications)
export function getSubscriber(): Redis {
  return new Redis(process.env.REDIS_URL);
}

import { Worker, Job } from "bullmq";
import { getBullMQConnection } from "../connection";

async function processChatJob(
  job: Job<ChatJobData, ChatJobResult>
): Promise<ChatJobResult> {
  const { userId, conversationId, messageId, message } = job.data;
  
  // Notify job started
  await notifyJobStarted(job.id!, conversationId, messageId);
  
  try {
    // Process message with AI agent
    const result = await processChatMessage({
      userId,
      conversationId,
      messageId,
      message,
    });
    
    // Notify completion
    await notifyJobCompleted(job.id!, conversationId, messageId);
    
    return result;
  } catch (error) {
    await notifyJobFailed(job.id!, conversationId, messageId, error);
    throw error;
  }
}

// Create worker
const worker = new Worker("chat", processChatJob, {
  connection: getBullMQConnection(),
  concurrency: parseInt(process.env.CHAT_QUEUE_CONCURRENCY || "5"),
  lockDuration: 180000, // 3 minutes
});

worker.on("completed", (job) => {
  logger.info({ jobId: job.id }, "chat_job_completed");
});

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