Decoupling AI: From Blocking Calls toEvent-Driven Microservices

import { Queue, Worker, QueueEvents } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis(process.env.REDIS_URL, {
  maxRetriesPerRequest: null, // Required by BullMQ
});

// The queue that holds pending AI jobs
export const aiJobQueue = new Queue('ai-jobs', { connection });

// Events emitter — used to listen for job completion
export const aiJobEvents = new QueueEvents('ai-jobs', { connection });

console.log('AI job queue initialized');

import express from 'express';
import { v4 as uuidv4 } from 'uuid';
import { aiJobQueue, aiJobEvents } from '../queue.js';

const router = express.Router();

router.post('/summarize', async (req, res) => {
  const { text, userId } = req.body;
  
  if (!text) {
    return res.status(400).json({ error: 'text is required' });
  }

  // Create a unique job ID the client will use to receive results
  const jobId = uuidv4();

  // Add to queue — this takes ~1ms and returns immediately
  await aiJobQueue.add(
    'summarize',
    { text, userId, jobId },
    {
      jobId,
      attempts: 3,           // Retry up to 3 times on failure
      backoff: {
        type: 'exponential', // Wait 2s, 4s, 8s between retries
        delay: 2000,
      },
    }
  );

  // Return immediately — the client will connect via WebSocket
  // to receive the result when the worker finishes
  return res.status(202).json({
    jobId,
    message: 'Job queued. Connect to WebSocket with this jobId to receive results.',
    wsEndpoint: `ws://your-domain/ws/jobs/${jobId}`
  });
});

export default router;

import asyncio
import json
import redis.asyncio as aioredis
from openai import AsyncOpenAI
import os
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

openai_client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
REDIS_URL = os.getenv("REDIS_URL")

async def process_ai_job(job_data: dict, redis_client) -> None:
    """Process a single AI job and publish the result."""
    job_id = job_data.get("jobId")
    text = job_data.get("text")
    
    logger.info(f"Processing job {job_id}")
    
    try:
        # Stream the LLM response token by token
        stream = await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": "You are an expert summarizer. Produce concise, accurate summaries."
                },
                {
                    "role": "user",
                    "content": f"Please summarize the following text:\n\n{text}"
                }
            ],
            stream=True,  # Stream tokens as they are generated
        )
        
        full_response = ""
        async for chunk in stream:
            token = chunk.choices[0].delta.content or ""
            full_response += token
            
            # Publish each token to the Redis channel in real-time
            # The Node.js WebSocket server is subscribed to this channel
            await redis_client.publish(
                f"job-result:{job_id}",
                json.dumps({"type": "token", "token": token, "jobId": job_id})
            )
        
        # Signal completion
        await redis_client.publish(
            f"job-result:{job_id}",
            json.dumps({"type": "complete", "result": full_response, "jobId": job_id})
        )
        
        logger.info(f"Job {job_id} completed successfully")
        
    except Exception as e:
        logger.error(f"Job {job_id} failed: {e}")
        await redis_client.publish(
            f"job-result:{job_id}",
            json.dumps({"type": "error", "error": str(e), "jobId": job_id})
        )

async def poll_queue(redis_client) -> None:
    """Poll the BullMQ Redis queue for new jobs."""
    # BullMQ stores jobs in a sorted set — we read directly from Redis
    # In production, use BullMQ's Python client or a dedicated queue reader
    logger.info("AI Worker started. Polling for jobs...")
    
    while True:
        # BLPOP blocks until a job is available, then returns it immediately
        result = await redis_client.blpop("bull:ai-jobs:wait", timeout=5)
        
        if result:
            _, job_key = result
            job_data_raw = await redis_client.hget(f"bull:ai-jobs:{job_key.decode()}", "data")
            
            if job_data_raw:
                job_data = json.loads(job_data_raw)
                await process_ai_job(job_data, redis_client)
        
        await asyncio.sleep(0.1)

async def main():
    redis_client = await aioredis.from_url(REDIS_URL, decode_responses=True)
    await poll_queue(redis_client)

if __name__ == "__main__":
    asyncio.run(main())

import { Server } from 'socket.io';
import { createClient } from 'redis';

export function initWebSocket(httpServer) {
  const io = new Server(httpServer, {
    cors: { origin: process.env.FRONTEND_URL }
  });

  const subscriber = createClient({ url: process.env.REDIS_URL });
  subscriber.connect();

  io.on('connection', (socket) => {
    const { jobId } = socket.handshake.query;
    
    if (!jobId) {
      socket.disconnect(true);
      return;
    }

    console.log(`Client connected, watching job: ${jobId}`);

    // Subscribe to this specific job's result channel
    const channel = `job-result:${jobId}`;
    
    subscriber.subscribe(channel, (message) => {
      const data = JSON.parse(message);
      
      if (data.type === 'token') {
        // Forward each token to the browser as it arrives
        socket.emit('ai:token', { token: data.token });
      } else if (data.type === 'complete') {
        socket.emit('ai:complete', { result: data.result });
        subscriber.unsubscribe(channel);
        socket.disconnect();
      } else if (data.type === 'error') {
        socket.emit('ai:error', { error: data.error });
        subscriber.unsubscribe(channel);
        socket.disconnect();
      }
    });

    socket.on('disconnect', () => {
      subscriber.unsubscribe(channel);
    });
  });
}