Transactional Outbox

A robust, dependency-injected Node.js implementation of the Transactional Outbox pattern for PostgreSQL.

⚠️ The Problem: "Dual Writes"

In distributed systems and microservices, a service often needs to update its own local database and simultaneously notify other services by publishing an event to a message broker (like Kafka or RabbitMQ).

If you try to perform these two actions sequentially without a distributed transaction coordinator, you encounter the Dual Write Problem:

1. If the database commit succeeds but the network call to Kafka fails, downstream services are never notified.
2. If Kafka succeeds but the database commit fails, downstream services react to "phantom" data that was rolled back.

💡 The Solution: Transactional Outbox

The Transactional Outbox pattern solves this by using your primary database as the single source of truth. Instead of sending the message to the broker directly, you insert the event into an outbox_events table within the exact same database transaction as your domain logic. A background worker then reliably relays those events to your message broker.

Architecture Diagram

  ┌───────────────────────┐
  │                       │ 1. start transaction
  │   Node.js Service     ├──────────┐
  │                       │          │
  └──────────┬────────────┘          │ 2. insert User
             │ 3. insert Event       │
             ▼                       ▼
 ┌──────────────────────────────────────┐                   ┌────────────────┐
 │  PostgreSQL Database                 │ 4. commit()       │                │
 │ ┌────────────┐ ┌───────────────────┐ │                   │                │
 │ │ users      │ │ outbox_events     │ │◄──────────────────┤  OutboxWorker  │
 │ └────────────┘ └─────────┬─────────┘ │ 5. poll pending   │                │
 └──────────────────────────┼───────────┘                   └────────┬───────┘
                            │                                        │
                            │                                        │ 6. publish()
                            │                                        ▼
                            │                               ┌────────────────┐
                            │ 7. mark done()                │ Message Broker │
                            └───────────────────────────────┤ (e.g., Kafka)  │
                                                            └────────────────┘

📦 Installation

npm install transactional-outbox pg uuid

(Note: pg and uuid are required peer/internal dependencies)

🚀 Quick Start & Usage Example

1. Database Schema

First, apply the schema.sql (found in the root of the repository) to your PostgreSQL database. This creates the required outbox_events table and the optimized idx_outbox_events_pending index for high-speed polling.

2. Service Integration

Here is a complete example of injecting the outbox into your transaction and starting the relay worker:

const { Pool } = require('pg');
const { 
  Outbox, 
  EventStore, 
  OutboxWorker, 
  KafkaPublisher 
} = require('transactional-outbox');

// 1. Initialize your PostgreSQL connection pool
const pool = new Pool({ connectionString: 'postgres://localhost/mydb' });

// 2. Setup the Outbox interceptor
const outbox = new Outbox(pool);

// 3. Setup the background Worker and your Broker Publisher
const eventStore = new EventStore(pool);
const publisher = new KafkaPublisher({ 
  clientId: 'my-service', 
  brokers: ['localhost:9092'] 
});

async function main() {
  // Connect cleanly to your message broker first
  await publisher.connect();
  
  // Start the background relay worker (polls every 2000ms by default)
  const worker = new OutboxWorker(eventStore, publisher);
  worker.start(2000);

  // ---------------------------------------------------------
  // 4. Safely wrap your domain logic and event in ONE transaction
  // ---------------------------------------------------------
  await outbox.publishTx(async (tx) => {
    
    // Standard domain mutation using the provided `tx` client
    const res = await tx.query(
      'INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id', 
      ['Alice', '[email protected]']
    );
    
    // Stage the event payload using the injected outbox helper
    await tx.outbox.publish('user.created', { 
      id: res.rows[0].id, 
      name: 'Alice' 
    });
    
  }); // <-- Automatically COMMITs both the user and the event!
  
  console.log('User created and event safely queued for background relay!');
}

main();

⚙️ Core Features

• Atomic Dual-Writes: Wraps your queries in a safe BEGIN/COMMIT block automatically.
• Concurrent-Safe Polling: Implements PostgreSQL's highly-optimized FOR UPDATE SKIP LOCKED query so multiple Node.js workers can poll the table simultaneously without database lock contention.
• At-Least-Once Delivery: Events are only marked as done after the broker successfully receives them.
• Exponential Backoff for Failures: Dead message broker? No problem. Built-in exponential backoff automatically schedules failing events for the future and takes permanently dead messages out of the loop after 5 retries.
• Pluggable Message Brokers: Ships with a ready-to-use KafkaPublisher and ConsolePublisher, but you can easily write your own RabbitMQ or SNS publisher as long as it exposes a simple publish(topic, payload) method.