1// BullMQ: Job queue with retries and DLQ
2 
3import { Queue, Worker } from 'bullmq';
4const emailQueue = new Queue('emails', { connection: redis });
5 
6// Producer: fire and forget in request handler
7export async function POST(req: Request) {
8  const order = await db.insert(orders).values(await req.json()).returning();
9 
10  // Queue email — do NOT await
Queue the email BEFORE returning response — do not await it
11  await emailQueue.add('order-confirmation', {
12    orderId: order[0].id,
13    email: order[0].userEmail,
14  }, {
15    attempts: 3,
16    backoff: { type: 'exponential', delay: 1000 },
17    removeOnComplete: 100,
18  });
19 
20  return Response.json({ orderId: order[0].id }, { status: 201 });
21  // Response in ~10ms; email sent asynchronously
22}
23 
24// Consumer: separate process
25const emailWorker = new Worker('emails', async (job) => {
26  await sendTransactionalEmail(job.data.email, job.data.orderId);
27}, { connection: redis, concurrency: 10 });
28 
29// Kafka: fanout to multiple consumer groups
30import { Kafka } from 'kafkajs';
31const producer = kafka.producer();
32 
33await producer.send({
Partition key guarantees ordering of events for the same order
34  topic: 'orders.placed',
35  messages: [{
36    key: orderId,  // Same key → same partition → ordered per-order
37    value: JSON.stringify({ orderId, userId, total }),
38  }],
39});
40// Analytics, Notifications, and Inventory services
41// each have their own consumer group — fully independent

1// Idempotent Consumer: safe to process the same message twice
2 
3async function processOrderPlaced(message: OrderPlacedEvent) {
4  const dedupKey = `processed:order:${message.orderId}`;
5 
6  // Check if already processed
7  if (await redis.exists(dedupKey)) {
8    console.log(`Skipping duplicate: ${message.orderId}`);
9    return; // ACK without processing
10  }
11 
12  await db.transaction(async (tx) => {
13    await tx.insert(orderAnalytics)
14      .values({ orderId: message.orderId, total: message.total })
onConflictDoNothing makes DB inserts idempotent at the DB level
15      .onConflictDoNothing();  // Safe to run twice — second run is a no-op
16  });
17 
Mark processed AFTER success — never before
18  // Mark processed AFTER successful processing
19  await redis.setex(dedupKey, 86400, '1'); // 24-hour dedup window
20}
21 
22// Outbox Pattern: atomically publish events with DB writes
23async function createOrderWithOutbox(orderData: OrderInput) {
24  await db.transaction(async (tx) => {
25    const order = await tx.insert(orders).values(orderData).returning();
26 
27    // Write event to outbox in SAME transaction as order creation
28    await tx.insert(outbox).values({
29      topic: 'orders.placed',
30      key: order[0].id,
31      payload: JSON.stringify({ orderId: order[0].id, ...orderData }),
32    });
33  });
34  // Outbox relay publishes to Kafka and marks as published
35  // If Kafka fails → retry from outbox → no event loss
36}