Outbox Pattern

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

// PRODUCTION-GRADE TRANSACTIONAL OUTBOX WRITE
 
import { v4 as uuidv4 } from 'uuid';
 
interface TransactionContext {
    query<T>(sql: string, params: unknown[]): Promise<T>;
    commit(): Promise<void>;
    rollback(): Promise<void>;
}
 
interface OutboxEvent {
    id: string;
    aggregateType: string;
    aggregateId: string;
    eventType: string;
    eventVersion: number;
    payload: unknown;
    metadata: EventMetadata;
}
 
interface EventMetadata {
    correlationId: string;
    causationId?: string;
    userId?: string;
    timestamp: Date;
}
 
class OrderService {
    constructor(private db: Database) {}
 
    async createOrder(
        command: CreateOrderCommand, 
        context: RequestContext
    ): Promise<Order> {
        // Generate IDs upfront - before transaction
        // This enables idempotency and correlation
        const orderId = uuidv4();
        const eventId = uuidv4();
 
        const order = await this.db.transaction(async (tx: TransactionContext) => {
            // === STEP 1: INSERT BUSINESS DATA ===
            
            const [order] = await tx.query<Order[]>(`
                INSERT INTO orders (
                    id, customer_id, status, 
                    shipping_address, billing_address,
                    subtotal, tax, shipping_cost, total,
                    created_at, updated_at
                ) VALUES (
                    $1, $2, $3, $4, $5, $6, $7, $8, $9, NOW(), NOW()
                )
                RETURNING *
            `, [
                orderId,
                command.customerId,
                'PENDING',
                JSON.stringify(command.shippingAddress),
                JSON.stringify(command.billingAddress),
                command.subtotal,
                command.tax,
                command.shippingCost,
                command.subtotal + command.tax + command.shippingCost
            ]);
 
            // === STEP 2: INSERT ORDER ITEMS ===
            
            for (const item of command.items) {
                await tx.query(`
                    INSERT INTO order_items (
                        id, order_id, product_id, 
                        quantity, unit_price, line_total
                    ) VALUES ($1, $2, $3, $4, $5, $6)
                `, [
                    uuidv4(),
                    orderId,
                    item.productId,
                    item.quantity,
                    item.unitPrice,
                    item.quantity * item.unitPrice
                ]);
            }
 
            // === STEP 3: INSERT OUTBOX EVENT (SAME TRANSACTION) ===
            
            const event: OutboxEvent = {
                id: eventId,
                aggregateType: 'Order',
                aggregateId: orderId,
                eventType: 'OrderCreated',
                eventVersion: 1,
                payload: {
                    orderId: orderId,
                    customerId: command.customerId,
                    items: command.items.map(i => ({
                        productId: i.productId,
                        productName: i.productName,
                        quantity: i.quantity,
                        unitPrice: i.unitPrice
                    })),
                    shippingAddress: command.shippingAddress,
                    subtotal: command.subtotal,
                    tax: command.tax,
                    shippingCost: command.shippingCost,
                    total: order.total,
                    currency: 'USD'
                },
                metadata: {
                    correlationId: context.correlationId,
                    causationId: context.requestId,
                    userId: context.userId,
                    timestamp: new Date()
                }
            };
 
            await tx.query(`
                INSERT INTO outbox_events (
                    id, aggregate_type, aggregate_id,
                    event_type, event_version,
                    payload, metadata,
                    correlation_id, causation_id,
                    created_at
                ) VALUES (
                    $1, $2, $3, $4, $5, $6, $7, $8, $9, NOW()
                )
            `, [
                event.id,
                event.aggregateType,
                event.aggregateId,
                event.eventType,
                event.eventVersion,
                JSON.stringify(event.payload),
                JSON.stringify(event.metadata),
                event.metadata.correlationId,
                event.metadata.causationId
            ]);
 
            return order;
        });
 
        // Transaction committed - order and event both exist
        return order;
    }
}
 
// KEY POINTS:
// 
// 1. IDs generated BEFORE transaction:
//    - Enables idempotency key usage
//    - Allows correlation before commit
//    - No need to retrieve database-generated IDs
//
// 2. All writes in SINGLE transaction:
//    - Order insert
//    - Order items inserts
//    - Outbox event insert
//    - All commit together or all rollback
//
// 3. Event payload constructed from SAME data:
//    - No chance of mismatch between order and event
//    - Event accurately reflects committed state
//
// 4. Metadata captured at creation time:
//    - Correlation ID for distributed tracing
//    - Causation ID links event to triggering action
//    - Timestamp is event creation time, not publish time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

-- SCENARIO: Transaction T1 inserts order + outbox event
-- Publisher P1 polls for unpublished events
-- What does P1 see?
 
-- === TIMELINE WITH READ COMMITTED ===
 
-- T=0: P1 starts polling (READ COMMITTED)
-- T=1: T1 begins transaction
-- T=2: T1 inserts order
-- T=3: T1 inserts outbox event
-- T=4: P1 queries outbox (sees NOTHING - T1 not committed)
-- T=5: T1 commits
-- T=6: P1 queries outbox (sees the event - T1 committed)
 
-- READ COMMITTED is SAFE - publisher never sees uncommitted events
 
 
-- === TIMELINE WITH REPEATABLE READ ===
 
-- T=0: P1 starts polling (REPEATABLE READ)
-- T=0: [P1's snapshot taken here]
-- T=1: T1 begins transaction
-- T=2: T1 inserts order
-- T=3: T1 inserts outbox event
-- T=4: T1 commits
-- T=5: P1 queries outbox (sees NOTHING despite T1 committed!)
-- T=6: P1's transaction ends
-- T=7: P1 starts new query (NOW sees the event)
 
-- REPEATABLE READ can DELAY visibility but is SAFE
-- Just need to start new transactions for each poll iteration
 
 
-- === TIMELINE WITH READ UNCOMMITTED (DANGEROUS!) ===
 
-- T=0: P1 starts polling (READ UNCOMMITTED)
-- T=1: T1 begins transaction
-- T=2: T1 inserts order
-- T=3: T1 inserts outbox event
-- T=4: P1 queries outbox (SEES uncommitted event!)
-- T=5: P1 publishes to message broker
-- T=6: T1 ROLLBACK (order never committed)
-- 
-- DISASTER: Event published for order that doesn't exist!
-- NEVER use READ UNCOMMITTED for outbox publisher
 
 
-- === RECOMMENDED: Explicit READ COMMITTED for publisher ===
 
-- PostgreSQL: Set per-transaction
BEGIN;
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT * FROM outbox_events 
WHERE published_at IS NULL 
ORDER BY sequence_number 
LIMIT 100
FOR UPDATE SKIP LOCKED;  -- Concurrent publisher safety
COMMIT;
 
-- MySQL: Set per-session or per-transaction
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
START TRANSACTION;
SELECT * FROM outbox_events 
WHERE published_at IS NULL 
ORDER BY sequence_number 
LIMIT 100
FOR UPDATE SKIP LOCKED;
COMMIT;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175

// CONCURRENT-SAFE OUTBOX PUBLISHER
 
class ConcurrentOutboxPublisher {
    constructor(
        private db: Database,
        private messageQueue: MessageQueue,
        private config: PublisherConfig
    ) {}
 
    async processBatch(): Promise<number> {
        // Use explicit READ COMMITTED for consistent behavior
        return await this.db.transaction(async (tx) => {
            // Set isolation level (PostgreSQL syntax)
            await tx.query('SET TRANSACTION ISOLATION LEVEL READ COMMITTED');
 
            // Select events with row-level locking
            // FOR UPDATE: Lock selected rows
            // SKIP LOCKED: Skip rows locked by other publishers
            const events = await tx.query<OutboxEvent[]>(`
                SELECT 
                    id, aggregate_type, aggregate_id,
                    event_type, event_version,
                    payload, metadata,
                    correlation_id, causation_id,
                    created_at, sequence_number
                FROM outbox_events
                WHERE published_at IS NULL
                ORDER BY sequence_number ASC
                LIMIT $1
                FOR UPDATE SKIP LOCKED
            `, [this.config.batchSize]);
 
            if (events.length === 0) {
                return 0;
            }
 
            // Track successfully published event IDs
            const publishedIds: string[] = [];
            const failedIds: string[] = [];
 
            for (const event of events) {
                try {
                    // Construct message for broker
                    const topic = this.getTopicName(event);
                    const message = {
                        key: event.aggregateId,  // Partition key for ordering
                        value: {
                            eventId: event.id,
                            eventType: event.eventType,
                            eventVersion: event.eventVersion,
                            aggregateType: event.aggregateType,
                            aggregateId: event.aggregateId,
                            payload: JSON.parse(event.payload),
                            metadata: JSON.parse(event.metadata),
                            timestamp: event.createdAt.toISOString()
                        },
                        headers: {
                            'correlation-id': event.correlationId,
                            'causation-id': event.causationId,
                            'event-type': event.eventType
                        }
                    };
 
                    // Publish with explicit acknowledgment
                    await this.messageQueue.publish(topic, message, {
                        acks: 'all',  // Wait for all replicas
                        timeout: this.config.publishTimeout
                    });
 
                    publishedIds.push(event.id);
                } catch (error) {
                    // Log failure but continue with other events
                    console.error(`Failed to publish event ${event.id}:`, error);
                    failedIds.push(event.id);
                    
                    // Update retry count for failed event
                    await tx.query(`
                        UPDATE outbox_events 
                        SET 
                            publish_attempts = publish_attempts + 1,
                            last_attempt_at = NOW(),
                            last_error = $1
                        WHERE id = $2
                    `, [String(error), event.id]);
                }
            }
 
            // Mark successfully published events
            if (publishedIds.length > 0) {
                await tx.query(`
                    UPDATE outbox_events 
                    SET published_at = NOW()
                    WHERE id = ANY($1)
                `, [publishedIds]);
            }
 
            return publishedIds.length;
        });
    }
 
    private getTopicName(event: OutboxEvent): string {
        // Convention: aggregate-type.event-type (lowercase, kebab-case)
        const aggregate = event.aggregateType.toLowerCase();
        const eventType = event.eventType
            .replace(/([A-Z])/g, '-$1')
            .toLowerCase()
            .replace(/^-/, '');
        return `${aggregate}.${eventType}`;
    }
}
 
// PUBLISHER ORCHESTRATION
 
class OutboxPublisherOrchestrator {
    private isRunning = false;
    private publishers: ConcurrentOutboxPublisher[] = [];
 
    constructor(
        private db: Database,
        private messageQueue: MessageQueue,
        private config: OrchestratorConfig
    ) {
        // Create publisher pool
        for (let i = 0; i < config.publisherCount; i++) {
            this.publishers.push(
                new ConcurrentOutboxPublisher(db, messageQueue, {
                    batchSize: config.batchSize,
                    publishTimeout: config.publishTimeout
                })
            );
        }
    }
 
    async start(): Promise<void> {
        this.isRunning = true;
 
        // Run all publishers concurrently
        await Promise.all(
            this.publishers.map((publisher, i) => 
                this.runPublisherLoop(publisher, i)
            )
        );
    }
 
    private async runPublisherLoop(
        publisher: ConcurrentOutboxPublisher,
        index: number
    ): Promise<void> {
        while (this.isRunning) {
            try {
                const processed = await publisher.processBatch();
 
                if (processed === 0) {
                    // No events found - back off to avoid spinning
                    await this.sleep(this.config.emptyPollDelay);
                } else {
                    // Events processed - immediately poll again
                    // (minimize latency for burst traffic)
                    await this.sleep(this.config.busyPollDelay);
                }
            } catch (error) {
                console.error(`Publisher ${index} error:`, error);
                await this.sleep(this.config.errorDelay);
            }
        }
    }
 
    private sleep(ms: number): Promise<void> {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
 
    async stop(): Promise<void> {
        this.isRunning = false;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

// STRATEGY 1: Claim Check Pattern
// ─────────────────────────────────
// Store large data in blob storage, put reference in event
 
interface ClaimCheckEvent {
    eventId: string;
    eventType: 'OrderDocumentGenerated';
    // Instead of embedding the 50MB PDF:
    documentRef: {
        storageType: 's3';
        bucket: 'order-documents';
        key: 'orders/ord-123/invoice.pdf';
        contentType: 'application/pdf';
        sizeBytes: 52428800;
        checksum: 'sha256:abc123...';
    };
    // Event is tiny; consumer fetches document via reference
}
 
class DocumentService {
    async generateInvoice(order: Order): Promise<void> {
        await this.db.transaction(async (tx) => {
            // Generate PDF (large)
            const pdfBuffer = await this.pdfGenerator.generateInvoice(order);
            
            // Store in blob storage (outside transaction is OK)
            const key = `orders/${order.id}/invoice.pdf`;
            await this.s3.putObject({
                Bucket: 'order-documents',
                Key: key,
                Body: pdfBuffer,
                ContentType: 'application/pdf'
            });
            
            // Store reference in outbox (small event)
            await tx.query(`
                INSERT INTO outbox_events (
                    id, aggregate_type, aggregate_id, event_type, payload
                ) VALUES ($1, $2, $3, $4, $5)
            `, [
                uuid(),
                'Order',
                order.id,
                'InvoiceGenerated',
                JSON.stringify({
                    orderId: order.id,
                    documentRef: {
                        storageType: 's3',
                        bucket: 'order-documents',
                        key: key,
                        contentType: 'application/pdf',
                        sizeBytes: pdfBuffer.length,
                        checksum: calculateSHA256(pdfBuffer)
                    }
                })
            ]);
        });
    }
}
 
 
// STRATEGY 2: Event Compression
// ─────────────────────────────
// Compress payload before storing
 
import { gzip, gunzip } from 'zlib';
import { promisify } from 'util';
 
const gzipAsync = promisify(gzip);
const gunzipAsync = promisify(gunzip);
 
class CompressedOutboxService {
    async addEvent(tx: Transaction, event: OutboxEvent): Promise<void> {
        const payloadJson = JSON.stringify(event.payload);
        
        // Compress if over threshold
        let storedPayload: Buffer | string;
        let isCompressed: boolean;
        
        if (payloadJson.length > 10_000) {  // 10KB threshold
            storedPayload = await gzipAsync(payloadJson);
            isCompressed = true;
        } else {
            storedPayload = payloadJson;
            isCompressed = false;
        }
        
        await tx.query(`
            INSERT INTO outbox_events (
                id, aggregate_type, aggregate_id, event_type,
                payload, is_compressed
            ) VALUES ($1, $2, $3, $4, $5, $6)
        `, [
            event.id,
            event.aggregateType,
            event.aggregateId,
            event.eventType,
            storedPayload,
            isCompressed
        ]);
    }
    
    async getEvent(id: string): Promise<OutboxEvent> {
        const row = await this.db.queryOne(`
            SELECT * FROM outbox_events WHERE id = $1
        `, [id]);
        
        let payload: object;
        if (row.is_compressed) {
            const decompressed = await gunzipAsync(row.payload);
            payload = JSON.parse(decompressed.toString());
        } else {
            payload = JSON.parse(row.payload);
        }
        
        return { ...row, payload };
    }
}
 
 
// STRATEGY 3: Event Splitting
// ───────────────────────────
// Split large event into header + multiple chunks
 
interface ChunkedEvent {
    eventId: string;
    chunkIndex: number;
    totalChunks: number;
    isLastChunk: boolean;
    payload: string;  // Partial payload
}
 
class ChunkedEventPublisher {
    private readonly CHUNK_SIZE = 500_000;  // 500KB chunks
    
    async publishLargeEvent(event: OutboxEvent): Promise<void> {
        const fullPayload = JSON.stringify(event.payload);
        
        if (fullPayload.length <= this.CHUNK_SIZE) {
            // Small enough - publish normally
            await this.messageQueue.publish(event);
            return;
        }
        
        // Split into chunks
        const totalChunks = Math.ceil(fullPayload.length / this.CHUNK_SIZE);
        
        for (let i = 0; i < totalChunks; i++) {
            const chunkPayload = fullPayload.slice(
                i * this.CHUNK_SIZE,
                (i + 1) * this.CHUNK_SIZE
            );
            
            const chunk: ChunkedEvent = {
                eventId: event.id,
                chunkIndex: i,
                totalChunks: totalChunks,
                isLastChunk: i === totalChunks - 1,
                payload: chunkPayload
            };
            
            await this.messageQueue.publish({
                topic: `${event.aggregateType.toLowerCase()}.chunked`,
                key: event.id,  // All chunks go to same partition
                value: chunk
            });
        }
    }
}
 
// Consumer reassembles chunks before processing