Pub/Sub Patterns

Core Concepts

Publisher: Produces events to a topic. Knows nothing about subscribers.

Topic: Named channel. Messages are sent to a topic, not to a specific subscriber.

Subscriber: Consumes events from one or more topics. Declares interest, receives matching events.

Consumer Group: Multiple instances of the same subscriber sharing the processing load. Each message is delivered to exactly one member of the group.

Topic: order-events
  ↓
┌─────────────────────────────────────┐
│     Consumer Group: notifications   │   ← one instance processes each message
│  [notification-service-1]           │
│  [notification-service-2]           │
│  [notification-service-3]           │
└─────────────────────────────────────┘
┌─────────────────────────────────────┐
│     Consumer Group: analytics       │   ← separate group gets its own copy
│  [analytics-service-1]              │
└─────────────────────────────────────┘

Two consumer groups on the same topic get independent copies of every message. Three instances within the same group share the load — each message goes to one of them.

Delivery Guarantees

Every pub/sub system makes a choice about what it guarantees:

At-most-once: Message delivered zero or one times. Can be lost. Fastest. Use for: metrics, telemetry, real-time dashboards where dropping a point is acceptable.

At-least-once: Message delivered one or more times. May be duplicated. Most common. Use for: anything that can be made idempotent (most business events).

Exactly-once: Message delivered exactly once. Most expensive. Use for: financial transactions, inventory deductions where duplicates cause real harm.

Most systems offer at-least-once and require consumers to handle deduplication:

async function handleOrderPlaced(event: EventEnvelope<OrderPlaced>): Promise<void> {
  // Idempotent: check if we already processed this event
  const alreadyProcessed = await db.processedEvents.exists(event.id);
  if (alreadyProcessed) {
    logger.info({ eventId: event.id }, 'Duplicate event, skipping');
    return;
  }

  await db.transaction(async (tx) => {
    // Process the event
    await tx.notifications.create({ userId: event.data.userId, type: 'order-placed' });

    // Mark as processed — atomic with the processing
    await tx.processedEvents.insert({ id: event.id, processedAt: new Date() });
  });
}

Fan-Out

One event → many subscribers, each doing different work:

OrderPlaced
  ├── notifications-service: send confirmation email
  ├── inventory-service: reserve items
  ├── analytics-service: update sales dashboard
  ├── fraud-service: check for suspicious patterns
  └── loyalty-service: award points

Each subscriber handles independently, fails independently, scales independently. Adding a new subscriber (e.g., a new loyalty program) requires zero changes to the order service.

Implementing fan-out with SNS + SQS (AWS):

import { SNS, SQS } from 'aws-sdk';

const sns = new SNS();
const sqs = new SQS();

// Publisher: sends to SNS topic
async function publishOrderPlaced(order: Order): Promise<void> {
  await sns.publish({
    TopicArn: process.env.ORDER_EVENTS_TOPIC_ARN!,
    Message: JSON.stringify({
      id: crypto.randomUUID(),
      type: 'OrderPlaced',
      version: 1,
      timestamp: new Date().toISOString(),
      data: {
        orderId: order.id,
        userId: order.userId,
        totalAmount: order.totalAmount,
      },
    }),
    MessageAttributes: {
      eventType: {
        DataType: 'String',
        StringValue: 'OrderPlaced',
      },
    },
  }).promise();
}

// Each subscriber has its own SQS queue subscribed to the SNS topic
// SNS automatically delivers to all subscribed queues
// Subscribers poll their own queue independently
async function processNotificationQueue(): Promise<void> {
  while (true) {
    const { Messages } = await sqs.receiveMessage({
      QueueUrl: process.env.NOTIFICATIONS_QUEUE_URL!,
      MaxNumberOfMessages: 10,
      WaitTimeSeconds: 20, // long polling
    }).promise();

    for (const message of Messages ?? []) {
      const event = JSON.parse(JSON.parse(message.Body!).Message);
      await handleOrderPlaced(event);

      await sqs.deleteMessage({
        QueueUrl: process.env.NOTIFICATIONS_QUEUE_URL!,
        ReceiptHandle: message.ReceiptHandle!,
      }).promise();
    }
  }
}

Message Filtering

Subscribers can filter to only receive events they care about — no need to receive and discard irrelevant events:

// SNS filter policy: only receive OrderPlaced with amount > $100
const filterPolicy = {
  eventType: ['OrderPlaced'],
  // Can't filter on nested fields with SNS filter policies directly
  // Use message attributes for filterable fields
};

// Publish with filterable attributes
await sns.publish({
  TopicArn: TOPIC_ARN,
  Message: JSON.stringify(event),
  MessageAttributes: {
    eventType: { DataType: 'String', StringValue: 'OrderPlaced' },
    orderAmount: { DataType: 'Number', StringValue: String(order.totalAmount) },
    plan: { DataType: 'String', StringValue: user.plan },
  },
}).promise();

// Subscription filter: VIP orders to a dedicated queue
// {
//   "plan": ["enterprise", "pro"],
//   "orderAmount": [{ "numeric": [">=", 1000] }]
// }

Kafka: Durable, Ordered, Replayable

Kafka is not just a message queue — it’s a persistent log. Messages are retained (configurable, often 7-30 days) and consumers can replay from any offset. This changes what’s possible:

import { Kafka, Consumer, Producer } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['kafka:9092'],
});

// Producer
const producer: Producer = kafka.producer();
await producer.connect();

await producer.send({
  topic: 'order-events',
  messages: [{
    key: order.userId,         // partition by user — ordering per user guaranteed
    value: JSON.stringify(event),
    headers: { eventType: 'OrderPlaced' },
  }],
});

// Consumer
const consumer: Consumer = kafka.consumer({ groupId: 'notifications-service' });
await consumer.connect();
await consumer.subscribe({ topic: 'order-events', fromBeginning: false });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value!.toString());

    try {
      await handleEvent(event);
      // Kafka commits offset after successful processing
      // On restart, consumer picks up from committed offset
    } catch (err) {
      // Don't commit — message will be redelivered
      logger.error({ event, err }, 'Failed to process event');
      throw err;
    }
  },
});

Kafka’s key properties:

• Ordering: Messages with the same partition key are strictly ordered
• Durability: Messages persisted to disk, replicated across brokers
• Replay: Consumers can seek to any offset and reprocess history
• Throughput: Millions of messages/second on modest hardware

When Kafka over SNS/SQS:

• Need message ordering within a partition
• Need to replay events (fix a bug in a consumer, reprocess historical data)
• Need to share events across teams/systems with different retention needs
• Throughput exceeds what managed queues handle economically

Ordering Guarantees

Ordering is only guaranteed within a partition (Kafka) or within a single FIFO queue (SQS FIFO). Cross-partition ordering is not guaranteed.

// Kafka: partition by user ID for per-user ordering
await producer.send({
  topic: 'user-events',
  messages: [{
    key: userId,   // all events for this user go to the same partition → ordered
    value: JSON.stringify(event),
  }],
});

// If you need global ordering: use a single partition
// Trade-off: single partition = single-threaded consumers = limited throughput

For most business events, per-entity ordering (all events for order_123 in order) is sufficient and achievable. Global ordering across all events usually isn’t needed and isn’t worth the throughput cost.

Dead-Letter Topics

Messages that fail processing after retries go to a dead-letter topic for investigation:

// Consumer with DLQ
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value!.toString());

    for (let attempt = 1; attempt <= 3; attempt++) {
      try {
        await handleEvent(event);
        return;
      } catch (err) {
        if (attempt === 3) {
          // Send to DLQ after 3 failures
          await dlqProducer.send({
            topic: `${topic}.dlq`,
            messages: [{
              key: message.key,
              value: message.value,
              headers: {
                ...message.headers,
                'x-original-topic': topic,
                'x-failure-reason': String(err),
                'x-failed-at': new Date().toISOString(),
              },
            }],
          });
          return; // don't rethrow — let consumer continue
        }
        await sleep(1000 * Math.pow(2, attempt));
      }
    }
  },
});

Monitor DLQ depth — a growing DLQ signals a consumer bug or a schema mismatch between producer and consumer.