Publish-Subscribe Pattern: Build Async Pipelines That Don't Fall Over at 3 AM

// io.thecodeforge — System Design tutorial

// Scenario: Order service publishes 'order_placed' event.
// Notification and Inventory services subscribe.

// Publisher (Order Service)
function publishOrderPlaced(orderId, userId, items) {
  const message = {
    event: 'order_placed',
    orderId: orderId,
    userId: userId,
    items: items,
    timestamp: Date.now()
  };
  // Publish to topic 'orders' — broker handles fan-out
  broker.publish('orders', JSON.stringify(message));
}

// Subscriber 1 (Notification Service)
broker.subscribe('orders', (message) => {
  const event = JSON.parse(message);
  if (event.event === 'order_placed') {
    sendEmail(event.userId, 'Your order is confirmed!');
  }
});

// Subscriber 2 (Inventory Service)
broker.subscribe('orders', (message) => {
  const event = JSON.parse(message);
  if (event.event === 'order_placed') {
    event.items.forEach(item => decrementStock(item.sku, item.quantity));
  }
});

// io.thecodeforge — System Design tutorial

// Kafka: durable, ordered, replayable
// Producer
producer.send(new ProducerRecord<>("orders", orderId, orderJson));
// Consumer (with offset management)
consumer.subscribe(Arrays.asList("orders"));
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  for (ConsumerRecord<String, String> record : records) {
    processOrder(record.value());
  }
  consumer.commitSync(); // ensures at-least-once
}

// Redis: fast, ephemeral, no persistence
// Publisher
redis.publish("orders", orderJson);
// Subscriber (blocking)
redis.subscribe(new JedisPubSub() {
  public void onMessage(String channel, String message) {
    processOrder(message);
  }
}, "orders");

// RabbitMQ: reliable, flexible routing
// Publisher
channel.basicPublish("orders_exchange", "order.placed", null, orderJson.getBytes());
// Consumer
channel.basicConsume("orders_queue", true, (consumerTag, delivery) -> {
  processOrder(new String(delivery.getBody()));
}, consumerTag -> {});

// io.thecodeforge — System Design tutorial

// Subscriber with bounded buffer and circuit breaker
const MAX_BUFFER_SIZE = 1000;
const processingQueue = [];
let isPaused = false;

broker.subscribe('orders', (message) => {
  if (processingQueue.length >= MAX_BUFFER_SIZE) {
    // Circuit breaker: pause subscription
    if (!isPaused) {
      broker.pause('orders');
      isPaused = true;
      alertOps('Order subscriber buffer full — pausing consumption');
    }
    // Reject message — broker will retry or DLQ
    return false;
  }
  processingQueue.push(message);
  processNext();
});

async function processNext() {
  if (processingQueue.length === 0) {
    if (isPaused) {
      broker.resume('orders');
      isPaused = false;
    }
    return;
  }
  const message = processingQueue.shift();
  try {
    await processOrder(message);
  } catch (err) {
    // Send to dead-letter queue
    deadLetterQueue.send(message);
  }
  setImmediate(processNext); // process next without stack overflow
}

// io.thecodeforge — System Design tutorial

// Idempotent subscriber using Redis dedup cache
const dedupCache = new RedisClient();
const DEDUP_TTL_SECONDS = 3600; // 1 hour — covers retry window

broker.subscribe('orders', async (message) => {
  const event = JSON.parse(message);
  const idempotencyKey = event.id; // e.g., 'order_placed:12345'
  
  // Check if already processed
  const alreadyProcessed = await dedupCache.get(idempotencyKey);
  if (alreadyProcessed) {
    logger.info(`Skipping duplicate message ${idempotencyKey}`);
    return;
  }
  
  // Process the message
  await processOrder(event);
  
  // Mark as processed
  await dedupCache.setex(idempotencyKey, DEDUP_TTL_SECONDS, '1');
});

async function processOrder(event) {
  // Database insert with ON CONFLICT DO NOTHING as safety net
  await db.query(`
    INSERT INTO orders (id, user_id, items, created_at)
    VALUES ($1, $2, $3, $4)
    ON CONFLICT (id) DO NOTHING
  `, [event.orderId, event.userId, JSON.stringify(event.items), new Date(event.timestamp)]);
}

// io.thecodeforge — System Design tutorial

// RabbitMQ: setup dead letter exchange
channel.exchangeDeclare("orders.dlx", "direct", true);
channel.queueDeclare("orders.dlq", true, false, false, null);
channel.queueBind("orders.dlq", "orders.dlx", "order.failed");

// Main queue with DLX config
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "orders.dlx");
args.put("x-dead-letter-routing-key", "order.failed");
args.put("x-max-retries", 3);
channel.queueDeclare("orders_queue", true, false, false, args);

// Consumer with retry count in headers
channel.basicConsume("orders_queue", false, (consumerTag, delivery) -> {
  try {
    processOrder(new String(delivery.getBody()));
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
  } catch (Exception e) {
    long retryCount = delivery.getProperties().getHeaders().getOrDefault("x-retry-count", 0);
    if (retryCount < 3) {
      // Reject and requeue with incremented retry count
      AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .headers(Map.of("x-retry-count", retryCount + 1))
        .build();
      channel.basicReject(delivery.getEnvelope().getDeliveryTag(), true);
      // Note: actual requeue with new headers requires custom logic
    } else {
      // Send to DLQ
      channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
    }
  }
}, consumerTag -> {});

// io.thecodeforge — System Design tutorial

// Scenario: User registration needs to send welcome email.
// Overkill: Kafka topic with one partition and one consumer
// Better: Direct HTTP call to email service with retry

async function registerUser(userData) {
  const user = await db.insertUser(userData);
  
  // Direct call with retry — simpler and synchronous
  const emailSent = await retry(() => 
    emailService.sendWelcomeEmail(user.email),
    { retries: 3, backoff: 'exponential' }
  );
  
  if (!emailSent) {
    // Fallback: queue for later processing
    await backgroundJobQueue.add('send_email', { userId: user.id });
  }
  
  return user;
}

// io.thecodeforge — System Design tutorial

// Publisher: attach trace ID
const traceId = uuidv4();
const message = {
  event: 'order_placed',
  orderId: orderId,
  traceId: traceId,
  timestamp: Date.now()
};
logger.info(`Publishing order_placed`, { traceId, orderId });
broker.publish('orders', JSON.stringify(message));

// Subscriber: log arrival and processing
broker.subscribe('orders', (rawMessage) => {
  const message = JSON.parse(rawMessage);
  const { traceId, orderId } = message;
  logger.info(`Received order_placed`, { traceId, orderId });
  
  const start = Date.now();
  try {
    processOrder(message);
    logger.info(`Processed order_placed`, { traceId, orderId, duration: Date.now() - start });
  } catch (err) {
    logger.error(`Failed to process order_placed`, { traceId, orderId, error: err.message });
  }
});