Outbox Pattern for Reliable Event Delivery

Overview

The Outbox Pattern solves a fundamental problem in distributed systems: how do you reliably publish events when business logic and event publishing must be atomic?

Without the Outbox Pattern:

• Business logic commits in database ✓
• Event publish to message broker fails ✗
• Result: Data inconsistency, lost events

With the Outbox Pattern:

• Business logic and event write to outbox happen in same database transaction
• Background worker reliably publishes events from outbox
• Result: At-least-once delivery guarantee

---

The Problem

Consider this booking confirmation flow:

// ❌ PROBLEMATIC: Not atomic
async confirmBooking(bookingId: string) {
  // Step 1: Update database (succeeds)
  await db.booking.update({
    where: { id: bookingId },
    data: { status: 'CONFIRMED' }
  });

  // Step 2: Publish event (fails - network issue)
  await messageBroker.publish({
    topic: 'booking.confirmed',
    payload: { bookingId }
  });  // ❌ Event lost! Database updated but no notification sent
}

Issues:

• If event publish fails, booking is confirmed but customers don't receive confirmation emails
• No automatic retry mechanism
• Manual intervention required to detect and fix lost events

---

The Solution

The Outbox Pattern guarantees atomic writes and eventual delivery:

// ✅ RELIABLE: Transactional event writes
async confirmBooking(bookingId: string) {
  await db.$transaction(async (tx) => {
    // Step 1: Update booking
    await tx.booking.update({
      where: { id: bookingId },
      data: { status: 'CONFIRMED' }
    });

    // Step 2: Write event to outbox (same transaction)
    await tx.messageOutbox.create({
      data: {
        id: uuid(),
        messageType: 'booking.confirmed',
        aggregateId: bookingId,
        aggregateType: 'booking',
        eventData: { bookingId, confirmedAt: new Date() },
        status: 'pending',
        tenantId: booking.tenantId
      }
    });
  });

  // Background worker will:
  // 1. Poll outbox for pending messages
  // 2. Publish to message broker/external systems
  // 3. Mark as 'sent' when successful
  // 4. Retry with exponential backoff if failed
}

Benefits:

• ✅ Atomicity: Event write is part of database transaction
• ✅ Reliability: At-least-once delivery guarantee
• ✅ Retry Logic: Automatic retry with exponential backoff
• ✅ Dead Letter Queue: Failed messages moved to DLQ after max retries
• ✅ Idempotency: Messages have unique IDs for deduplication

---

Database Schema

MessageOutbox Table

Location: packages/db/prisma/schema.prisma

model MessageOutbox {
  id            String   @id
  tenantId      String
  messageType   String   // e.g., "booking.confirmed", "payment.processed"
  aggregateId   String   // e.g., Booking ID, Payment ID
  aggregateType String   // e.g., "booking", "payment", "promotion"
  eventVersion  Int      @default(1)  // Event schema version
  eventData     Json     // Complete event payload
  metadata      Json?    @default("{}")  // Correlation IDs, trace context
  status        String   @default("pending")  // pending, processing, sent, failed, dead_letter
  scheduledAt   DateTime @default(now())  // When to process (for delayed events)
  processedAt   DateTime?
  retryCount    Int      @default(0)
  maxRetries    Int      @default(3)
  nextRetryAt   DateTime?
  lastError     String?
  createdAt     DateTime @default(now())
  updatedAt     DateTime @updatedAt

  @@index([status, scheduledAt])  // Query pending messages efficiently
  @@index([tenantId, aggregateType, aggregateId])  // Query events by aggregate
  @@index([messageType, status])  // Monitor specific message types
  @@map("message_outbox")
}

Key Fields:

• id: Unique message identifier (UUID) for idempotency
• messageType: Defines the event type for routing
• aggregateId/Type: Links event to domain entity
• eventData: Complete event payload as JSON
• metadata: Correlation IDs, saga IDs, trace context
• status: Message lifecycle state
• scheduledAt: Supports delayed event processing
• retryCount/maxRetries: Retry logic parameters
• nextRetryAt: Exponential backoff scheduling

---

Message Lifecycle

Status Flow

pending → processing → sent ✓
             ↓
          failed → (retry) → processing → sent ✓
             ↓
          dead_letter ✗ (max retries exceeded)

Status Definitions:

Status	Description	Next States
pending	Event created, waiting for processing	processing
processing	Worker is actively publishing event	sent, failed
sent	Event successfully delivered	Terminal state ✓
failed	Publish attempt failed, retry pending	processing (retry), dead_letter (max retries)
dead_letter	Max retries exceeded, manual review needed	Terminal state ✗

Event Processing Flow

1. Business Transaction Commits
       ↓
   Event in outbox (status: pending)
       ↓
2. Worker Polls Outbox
       ↓
   Update status to 'processing'
       ↓
3. Attempt Delivery
   ├─ Success → status = 'sent', processedAt = now()
   └─ Failure → status = 'failed', nextRetryAt = calculated backoff
       ↓
4. Retry Logic
   ├─ retryCount < maxRetries → Retry with backoff
   └─ retryCount >= maxRetries → status = 'dead_letter'

---

Implementation

Service: OutboxProducerService

Location: apps/api/src/modules/notifications/outbox-producer.service.ts

Writes events to outbox within business transactions:

@Injectable()
export class OutboxProducerService {
  // Publish event to outbox (called within transaction)
  async publishEvent(params: PublishEventParams, tx?: PrismaTransaction) {
    const dbClient = tx || this.db;

    const message = await dbClient.messageOutbox.create({
      data: {
        id: uuid(),
        tenantId: params.tenantId,
        messageType: params.messageType,
        aggregateId: params.aggregateId,
        aggregateType: params.aggregateType,
        eventData: params.eventData,
        metadata: {
          correlationId: params.correlationId,
          sagaId: params.sagaId,
          timestamp: new Date().toISOString(),
        },
        status: 'pending',
        scheduledAt: params.scheduledAt || new Date(),
        maxRetries: params.maxRetries || 3,
      },
    });

    this.logger.info('Event published to outbox', {
      messageId: message.id,
      messageType: message.messageType,
      aggregateId: message.aggregateId,
    });

    return message;
  }

  // Publish batch of events (transactional)
  async publishBatch(events: PublishEventParams[], tx?: PrismaTransaction) {
    const dbClient = tx || this.db;

    const messages = events.map((event) => ({
      id: uuid(),
      tenantId: event.tenantId,
      messageType: event.messageType,
      aggregateId: event.aggregateId,
      aggregateType: event.aggregateType,
      eventData: event.eventData,
      metadata: event.metadata || {},
      status: 'pending',
      scheduledAt: event.scheduledAt || new Date(),
      maxRetries: event.maxRetries || 3,
    }));

    await dbClient.messageOutbox.createMany({
      data: messages,
    });

    return messages;
  }

  // Schedule delayed event
  async scheduleEvent(params: PublishEventParams, delayMs: number) {
    const scheduledAt = new Date(Date.now() + delayMs);

    return this.publishEvent({
      ...params,
      scheduledAt,
    });
  }
}

Service: OutboxWorkerService

Location: apps/api/src/modules/notifications/outbox-worker.service.ts

Background worker that processes outbox messages:

@Injectable()
export class OutboxWorkerService implements OnModuleInit {
  private readonly BATCH_SIZE = 50;
  private readonly POLL_INTERVAL = 1000; // 1 second
  private readonly PROCESSING_TIMEOUT = 30000; // 30 seconds
  private isRunning = false;

  async onModuleInit() {
    // Start worker on module initialization
    await this.startWorker();
  }

  async startWorker() {
    if (this.isRunning) return;

    this.isRunning = true;
    this.logger.info('Outbox worker started');

    // Continuous polling loop
    while (this.isRunning) {
      try {
        await this.processBatch();
      } catch (error) {
        this.logger.error('Error processing outbox batch', error);
      }

      // Wait before next poll
      await this.sleep(this.POLL_INTERVAL);
    }
  }

  async processBatch() {
    const now = new Date();

    // Find pending messages ready to process
    const messages = await this.db.messageOutbox.findMany({
      where: {
        status: 'pending',
        scheduledAt: { lte: now },
      },
      orderBy: { scheduledAt: 'asc' },
      take: this.BATCH_SIZE,
    });

    if (messages.length === 0) {
      // Also check for failed messages ready to retry
      await this.retryFailedMessages(now);
      return;
    }

    this.logger.info(`Processing ${messages.length} outbox messages`);

    // Process messages in parallel (with concurrency limit)
    await Promise.all(messages.map((message) => this.processMessage(message)));
  }

  async processMessage(message: MessageOutbox) {
    try {
      // Mark as processing
      await this.db.messageOutbox.update({
        where: { id: message.id },
        data: { status: 'processing' },
      });

      // Route message based on type
      await this.deliverMessage(message);

      // Mark as sent
      await this.db.messageOutbox.update({
        where: { id: message.id },
        data: {
          status: 'sent',
          processedAt: new Date(),
        },
      });

      this.logger.info('Message delivered successfully', {
        messageId: message.id,
        messageType: message.messageType,
      });
    } catch (error) {
      await this.handleFailure(message, error);
    }
  }

  async deliverMessage(message: MessageOutbox) {
    // Route message based on type
    const handler = this.getMessageHandler(message.messageType);

    if (!handler) {
      throw new Error(`No handler for message type: ${message.messageType}`);
    }

    // Execute handler with timeout
    await Promise.race([
      handler.handle(message),
      this.timeout(this.PROCESSING_TIMEOUT),
    ]);
  }

  async handleFailure(message: MessageOutbox, error: Error) {
    const retryCount = message.retryCount + 1;

    if (retryCount >= message.maxRetries) {
      // Max retries exceeded - move to dead letter queue
      await this.db.messageOutbox.update({
        where: { id: message.id },
        data: {
          status: 'dead_letter',
          retryCount,
          lastError: error.message,
          processedAt: new Date(),
        },
      });

      this.logger.error('Message moved to dead letter queue', {
        messageId: message.id,
        messageType: message.messageType,
        error: error.message,
      });

      // Alert on dead letter queue
      await this.alertService.notify({
        severity: 'high',
        title: 'Message in Dead Letter Queue',
        message: `Message ${message.id} failed after ${retryCount} attempts`,
      });
    } else {
      // Calculate next retry with exponential backoff
      const nextRetryAt = this.calculateNextRetry(retryCount);

      await this.db.messageOutbox.update({
        where: { id: message.id },
        data: {
          status: 'failed',
          retryCount,
          nextRetryAt,
          lastError: error.message,
        },
      });

      this.logger.warn('Message failed, scheduled for retry', {
        messageId: message.id,
        retryCount,
        nextRetryAt,
      });
    }
  }

  async retryFailedMessages(now: Date) {
    // Find failed messages ready to retry
    const messagesToRetry = await this.db.messageOutbox.findMany({
      where: {
        status: 'failed',
        nextRetryAt: { lte: now },
      },
      take: this.BATCH_SIZE,
    });

    // Reset to pending for processing
    await this.db.messageOutbox.updateMany({
      where: {
        id: { in: messagesToRetry.map((m) => m.id) },
      },
      data: { status: 'pending' },
    });
  }

  // Exponential backoff: 1s, 2s, 4s, 8s, etc.
  private calculateNextRetry(retryCount: number): Date {
    const baseDelay = 1000; // 1 second
    const delay = baseDelay * Math.pow(2, retryCount);
    const jitter = Math.random() * 1000; // Add jitter

    return new Date(Date.now() + delay + jitter);
  }

  private timeout(ms: number): Promise<never> {
    return new Promise((_, reject) =>
      setTimeout(() => reject(new Error('Timeout')), ms),
    );
  }

  private sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }
}

---

Message Handlers

Email Handler

@Injectable()
export class EmailMessageHandler implements MessageHandler {
  canHandle(messageType: string): boolean {
    return messageType.startsWith('email.');
  }

  async handle(message: MessageOutbox) {
    const { messageType, eventData } = message;

    switch (messageType) {
      case 'email.booking_confirmation':
        await this.sendBookingConfirmation(eventData);
        break;

      case 'email.booking_reminder':
        await this.sendBookingReminder(eventData);
        break;

      case 'email.booking_cancellation':
        await this.sendCancellationNotice(eventData);
        break;

      default:
        throw new Error(`Unknown email type: ${messageType}`);
    }
  }

  private async sendBookingConfirmation(data: any) {
    await this.emailService.send({
      to: data.customerEmail,
      template: 'booking-confirmation',
      context: {
        bookingId: data.bookingId,
        customerName: data.customerName,
        bookingDate: data.startsAt,
        product: data.productName,
      },
    });
  }
}

Webhook Handler

@Injectable()
export class WebhookMessageHandler implements MessageHandler {
  canHandle(messageType: string): boolean {
    return messageType.startsWith('webhook.');
  }

  async handle(message: MessageOutbox) {
    // Get tenant webhook configuration
    const tenant = await this.db.tenant.findUnique({
      where: { id: message.tenantId },
    });

    if (!tenant.webhookUrl) {
      this.logger.warn('Tenant has no webhook URL configured', {
        tenantId: message.tenantId,
      });
      return;
    }

    // Compute signature for webhook security
    const signature = this.computeSignature(
      message.eventData,
      tenant.webhookSecret,
    );

    // POST event to tenant webhook
    const response = await fetch(tenant.webhookUrl, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-Event-Type': message.messageType,
        'X-Event-ID': message.id,
        'X-Signature': signature,
      },
      body: JSON.stringify(message.eventData),
    });

    if (!response.ok) {
      throw new Error(
        `Webhook failed: ${response.status} ${response.statusText}`,
      );
    }
  }

  private computeSignature(payload: any, secret: string): string {
    const hmac = crypto.createHmac('sha256', secret);
    hmac.update(JSON.stringify(payload));
    return hmac.digest('hex');
  }
}

---

Common Message Types

Booking Events

// Booking created (hold placed)
{
  messageType: 'booking.created',
  aggregateId: bookingId,
  aggregateType: 'booking',
  eventData: {
    bookingId,
    customerId,
    productId,
    startsAt,
    endsAt,
    partySize,
    status: 'HOLD'
  }
}

// Booking confirmed (payment captured)
{
  messageType: 'booking.confirmed',
  aggregateId: bookingId,
  aggregateType: 'booking',
  eventData: {
    bookingId,
    confirmedAt: new Date(),
    paymentId,
    amountPaid
  }
}

// Booking cancelled
{
  messageType: 'booking.cancelled',
  aggregateId: bookingId,
  aggregateType: 'booking',
  eventData: {
    bookingId,
    cancelledAt: new Date(),
    cancelledBy: 'customer',  // or 'admin', 'system'
    refundAmount
  }
}

Payment Events

// Payment processed
{
  messageType: 'payment.processed',
  aggregateId: paymentId,
  aggregateType: 'payment',
  eventData: {
    paymentId,
    bookingId,
    amount,
    currency,
    provider: 'stripe',
    status: 'captured'
  }
}

// Refund processed
{
  messageType: 'payment.refunded',
  aggregateId: refundId,
  aggregateType: 'payment',
  eventData: {
    refundId,
    paymentId,
    bookingId,
    refundAmount,
    feeDeducted,
    netRefund
  }
}

Email Events

// Booking confirmation email
{
  messageType: 'email.booking_confirmation',
  aggregateId: bookingId,
  aggregateType: 'booking',
  eventData: {
    to: 'customer@example.com',
    bookingId,
    customerName,
    startsAt,
    productName
  }
}

---

Best Practices

1. Always Use Transactions

Ensure business logic and event writes are atomic:

// ✅ GOOD: Transactional event write
async function confirmBooking(bookingId: string) {
  await db.$transaction(async (tx) => {
    // Business logic
    await tx.booking.update({
      where: { id: bookingId },
      data: { status: 'CONFIRMED' },
    });

    // Event write (same transaction)
    await outboxProducer.publishEvent(
      {
        messageType: 'booking.confirmed',
        aggregateId: bookingId,
        // ...
      },
      tx,
    ); // ✓ Pass transaction
  });
}

// ❌ BAD: Not transactional
async function confirmBooking(bookingId: string) {
  await db.booking.update({
    /* ... */
  }); // ❌ Separate transaction
  await outboxProducer.publishEvent({
    /* ... */
  }); // ❌ Can fail independently
}

2. Include Correlation IDs

Link events to sagas and distributed traces:

await outboxProducer.publishEvent({
  messageType: 'booking.confirmed',
  aggregateId: bookingId,
  eventData: {
    /* ... */
  },
  metadata: {
    correlationId: saga.correlationId, // ✓ Link to saga
    sagaId: saga.id,
    traceId: context.traceId, // ✓ Distributed tracing
    userId: context.userId,
  },
});

3. Use Event Versioning

Support schema evolution:

// Version 1
{
  eventVersion: 1,
  eventData: {
    bookingId,
    startsAt
  }
}

// Version 2 (added participant breakdown)
{
  eventVersion: 2,
  eventData: {
    bookingId,
    startsAt,
    participantTypes: [{ typeId: 'adult', count: 2 }]
  }
}

// Handler supports multiple versions
async handle(message: MessageOutbox) {
  if (message.eventVersion === 1) {
    return this.handleV1(message.eventData);
  } else if (message.eventVersion === 2) {
    return this.handleV2(message.eventData);
  }
}

4. Implement Idempotent Handlers

Handlers must be safe to retry:

// ✅ GOOD: Idempotent email handler
async sendEmail(message: MessageOutbox) {
  // Check if already sent (using message ID)
  const existing = await this.emailLog.findUnique({
    where: { messageId: message.id }
  });

  if (existing) {
    this.logger.info('Email already sent, skipping');
    return;  // ✓ Idempotent
  }

  // Send email
  await this.emailService.send(/* ... */);

  // Record sent
  await this.emailLog.create({
    data: { messageId: message.id, sentAt: new Date() }
  });
}

5. Monitor Dead Letter Queue

Alert on messages requiring manual intervention:

// Daily DLQ check
@Cron('0 9 * * *')  // 9 AM daily
async checkDeadLetterQueue() {
  const dlqCount = await this.db.messageOutbox.count({
    where: { status: 'dead_letter' }
  });

  if (dlqCount > 0) {
    await this.alertService.notify({
      severity: 'high',
      title: 'Messages in Dead Letter Queue',
      message: `${dlqCount} messages require manual review`,
      link: '/admin/outbox/dead-letter'
    });
  }
}

---

Monitoring and Observability

Key Metrics

// Message throughput
const throughput = await db.messageOutbox.count({
  where: {
    status: 'sent',
    processedAt: { gte: last1Hour },
  },
});

// Processing latency
const avgLatency = await db.$queryRaw`
  SELECT
    AVG(EXTRACT(EPOCH FROM (processed_at - created_at))) as avg_latency_seconds
  FROM message_outbox
  WHERE status = 'sent'
    AND processed_at >= NOW() - INTERVAL '1 hour'
`;

// Failure rate
const failureRate =
  (await db.messageOutbox.count({
    where: {
      status: { in: ['failed', 'dead_letter'] },
      createdAt: { gte: last24Hours },
    },
  })) / totalMessages;

// Retry distribution
const retryDistribution = await db.messageOutbox.groupBy({
  by: ['retryCount'],
  _count: true,
  where: {
    status: 'sent',
    retryCount: { gt: 0 },
  },
});

Dashboards

Monitor outbox health in real-time:

• Message Throughput: Messages processed per minute
• Processing Latency: Time from creation to delivery
• Failure Rate: % of messages failing
• Retry Distribution: How often retries are needed
• Dead Letter Queue Size: Messages requiring intervention
• Message Age: Oldest pending message age

---

Troubleshooting

Messages Not Processing

Symptoms: Messages stuck in pending status

Causes:

• Worker not running
• Database connection issues
• scheduledAt set to future date

Resolution:

-- Check pending message count
SELECT COUNT(*) FROM message_outbox WHERE status = 'pending';

-- Check oldest pending message
SELECT * FROM message_outbox
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT 1;

-- Manually trigger processing (if worker down)
UPDATE message_outbox
SET status = 'pending', next_retry_at = NOW()
WHERE status = 'processing'
  AND updated_at < NOW() - INTERVAL '5 minutes';

High Dead Letter Queue Count

Symptoms: Many messages with status = 'dead_letter'

Causes:

• External service consistently failing
• Invalid event data
• Handler bugs

Resolution:

1. Review error messages:

SELECT message_type, last_error, COUNT(*)
FROM message_outbox
WHERE status = 'dead_letter'
GROUP BY message_type, last_error;

2. Fix underlying issue (service, data, code)

3. Retry messages:

-- Reset to pending for reprocessing
UPDATE message_outbox
SET status = 'pending',
    retry_count = 0,
    next_retry_at = NULL,
    last_error = NULL
WHERE id IN (
  SELECT id FROM message_outbox
  WHERE status = 'dead_letter'
    AND message_type = 'email.booking_confirmation'
);

---

Related Documentation

• Saga Orchestration - Distributed transaction coordination
• Database Schema - Outbox table structure
• Payments API - Payment event examples

---

Summary

The Outbox Pattern enables:

• ✅ At-least-once delivery guarantees
• ✅ Atomic event writes with business logic
• ✅ Automatic retry with exponential backoff
• ✅ Dead Letter Queue for failed messages
• ✅ Idempotent message processing
• ✅ Correlation tracking across services

Use the Outbox Pattern whenever you need reliable event delivery from your database to external systems.