Event-Driven Integration

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

// DANGEROUS: Dual-write with no atomicity guarantee
class OrderService {
    async placeOrder(command: PlaceOrderCommand): Promise<Order> {
        // Step 1: Save to database
        const order = Order.create(command);
        await this.orderRepository.save(order);  // ← Commits
        
        // Step 2: Publish event  
        await this.eventBus.publish(
            new OrderPlacedEvent(order)
        );  // ← What if this fails?
        
        return order;
    }
}
 
// FAILURE SCENARIO 1: Event publish fails
// - Database has the order (committed)
// - Event never sent
// - Other services never know about the order
// - System is inconsistent!
 
// FAILURE SCENARIO 2: Service crashes after DB commit
// - Order is saved
// - Service crashes before publish
// - Event lost forever
// - Same inconsistency!
 
// WRONG FIX: Publish first, then save
async placeOrderWrongFix(command: PlaceOrderCommand): Promise<Order> {
    const order = Order.create(command);
    
    // Publish first
    await this.eventBus.publish(new OrderPlacedEvent(order));
    
    // Then save
    await this.orderRepository.save(order);  // ← What if THIS fails?
    
    // Now: Event was sent, but order wasn't saved!
    // Other services act on non-existent order!
}

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

// Outbox table entity
interface OutboxMessage {
    id: string;              // Unique message ID
    aggregateType: string;   // e.g., "Order"
    aggregateId: string;     // e.g., order ID
    eventType: string;       // e.g., "order.placed"
    payload: string;         // JSON serialized event
    createdAt: Date;         // When event was created
    publishedAt: Date | null; // When event was published (null = pending)
}
 
// Service writes to outbox in same transaction
class OrderService {
    constructor(
        private db: TransactionManager,
        private orderRepository: OrderRepository,
        private outboxRepository: OutboxRepository,
    ) {}
    
    async placeOrder(command: PlaceOrderCommand): Promise<Order> {
        // Single database transaction wraps everything
        return await this.db.transaction(async (tx) => {
            // Create and save order
            const order = Order.create(command);
            await this.orderRepository.save(order, tx);
            
            // Instead of publishing, write to outbox
            const event = this.createOrderPlacedEvent(order);
            await this.outboxRepository.insert({
                id: generateUuid(),
                aggregateType: "Order",
                aggregateId: order.id,
                eventType: "order.placed",
                payload: JSON.stringify(event),
                createdAt: new Date(),
                publishedAt: null,  // Pending
            }, tx);
            
            return order;
            // Transaction commits: order AND outbox message are atomic
        });
    }
}
 
// Separate relay process publishes from outbox
class OutboxRelay {
    constructor(
        private outboxRepository: OutboxRepository,
        private eventBus: EventBus,
    ) {}
    
    async run(): Promise<void> {
        // Poll for unpublished messages
        while (true) {
            const messages = await this.outboxRepository.findUnpublished(
                { limit: 100 }
            );
            
            for (const message of messages) {
                try {
                    // Publish to message broker
                    await this.eventBus.publish({
                        eventId: message.id,
                        eventType: message.eventType,
                        payload: JSON.parse(message.payload),
                    });
                    
                    // Mark as published
                    await this.outboxRepository.markPublished(message.id);
                    
                } catch (error) {
                    // Will retry on next poll
                    console.error(`Failed to publish ${message.id}`, error);
                }
            }
            
            // Brief pause before next poll
            await sleep(100);
        }
    }
}

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

-- Outbox table design
CREATE TABLE outbox_messages (
    id UUID PRIMARY KEY,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    published_at TIMESTAMP WITH TIME ZONE,
    
    -- Index for polling unpublished messages efficiently
    -- Partial index only includes unpublished rows
    CONSTRAINT outbox_unpublished_idx 
        ON outbox_messages (created_at) 
        WHERE published_at IS NULL
);
 
-- Alternative: Use a sequence number for ordering
CREATE TABLE outbox_messages_v2 (
    sequence_number BIGSERIAL PRIMARY KEY,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
-- Relay tracks last processed sequence_number
-- No need for published_at column

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

// CHOREOGRAPHY: Each service reacts to events
 
// Step 1: Order Service creates order, publishes event
class OrderService {
    async createOrder(command: CreateOrderCommand): Promise<void> {
        const order = Order.createPending(command);
        await this.orderRepository.save(order);
        await this.publish(new OrderCreatedEvent(order));
        // Now waiting for payment processing...
    }
    
    // Listens for payment results
    async onPaymentCompleted(event: PaymentCompletedEvent): Promise<void> {
        const order = await this.orderRepository.get(event.orderId);
        order.markPaymentReceived();
        await this.orderRepository.save(order);
        // Inventory service listens and reacts...
    }
    
    async onPaymentFailed(event: PaymentFailedEvent): Promise<void> {
        const order = await this.orderRepository.get(event.orderId);
        order.cancel("Payment failed");
        await this.orderRepository.save(order);
        await this.publish(new OrderCancelledEvent(order));
    }
}
 
// Step 2: Payment Service listens for orders
class PaymentService {
    async onOrderCreated(event: OrderCreatedEvent): Promise<void> {
        try {
            const result = await this.processPayment(
                event.customerId,
                event.totalAmount
            );
            await this.publish(new PaymentCompletedEvent({
                orderId: event.orderId,
                paymentId: result.id,
            }));
        } catch (error) {
            await this.publish(new PaymentFailedEvent({
                orderId: event.orderId,
                reason: error.message,
            }));
        }
    }
    
    // Compensating transaction
    async onOrderCancelled(event: OrderCancelledEvent): Promise<void> {
        const payment = await this.paymentRepository.findByOrderId(
            event.orderId
        );
        if (payment) {
            await this.refundPayment(payment);
            await this.publish(new PaymentRefundedEvent({
                orderId: event.orderId,
                paymentId: payment.id,
            }));
        }
    }
}
 
// Step 3: Inventory Service listens for payment
class InventoryService {
    async onPaymentCompleted(event: PaymentCompletedEvent): Promise<void> {
        try {
            await this.reserveInventory(event.orderId);
            await this.publish(new InventoryReservedEvent({
                orderId: event.orderId,
            }));
        } catch (error) {
            // Trigger compensation: refund needed
            await this.publish(new InventoryReservationFailedEvent({
                orderId: event.orderId,
                reason: error.message,
            }));
        }
    }
    
    // Compensating transaction
    async onOrderCancelled(event: OrderCancelledEvent): Promise<void> {
        await this.releaseReservation(event.orderId);
    }
}

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

// ORCHESTRATION: Central coordinator manages the saga
 
interface SagaStep {
    readonly name: string;
    execute(context: SagaContext): Promise<void>;
    compensate(context: SagaContext): Promise<void>;
}
 
class OrderFulfillmentSaga {
    private readonly steps: SagaStep[] = [
        new CreateOrderStep(),
        new ProcessPaymentStep(),
        new ReserveInventoryStep(),
        new CreateShipmentStep(),
    ];
    
    async execute(command: CreateOrderCommand): Promise<void> {
        const context = new SagaContext(command);
        const completedSteps: SagaStep[] = [];
        
        try {
            for (const step of this.steps) {
                await step.execute(context);
                completedSteps.push(step);
                
                // Persist saga state for recovery
                await this.saveSagaState(context, completedSteps);
            }
            
            context.markCompleted();
            await this.saveSagaState(context, completedSteps);
            
        } catch (error) {
            // Compensate in reverse order
            context.markFailed(error);
            
            for (const step of completedSteps.reverse()) {
                try {
                    await step.compensate(context);
                } catch (compensationError) {
                    // Log but continue compensating
                    console.error(
                        `Compensation failed for ${step.name}`,
                        compensationError
                    );
                }
            }
            
            await this.saveSagaState(context, completedSteps);
            throw error;
        }
    }
}
 
// Each step has execute and compensate logic
class ProcessPaymentStep implements SagaStep {
    readonly name = "ProcessPayment";
    
    async execute(context: SagaContext): Promise<void> {
        const result = await this.paymentService.processPayment({
            orderId: context.orderId,
            amount: context.totalAmount,
            customerId: context.customerId,
        });
        
        context.setPaymentId(result.paymentId);
    }
    
    async compensate(context: SagaContext): Promise<void> {
        if (context.paymentId) {
            await this.paymentService.refundPayment(context.paymentId);
        }
    }
}
 
// Saga orchestrator runs as a separate service
class SagaOrchestrator {
    async start(): Promise<void> {
        // Listen for saga initiation requests
        await this.eventBus.subscribe(
            "order.create.requested",
            async (event) => {
                const saga = new OrderFulfillmentSaga();
                await saga.execute(event.command);
            }
        );
        
        // Handle saga recovery on startup
        await this.recoverIncompleteSagas();
    }
    
    private async recoverIncompleteSagas(): Promise<void> {
        const incompleteSagas = await this.sagaRepository.findIncomplete();
        for (const sagaState of incompleteSagas) {
            await this.resumeOrCompensate(sagaState);
        }
    }
}

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

// POINT-TO-POINT: Queue with exactly-once delivery
class TaskQueue {
    async enqueue(task: Task): Promise<void> {
        await this.redis.lpush("tasks", JSON.stringify(task));
    }
    
    async dequeue(): Promise<Task | null> {
        const data = await this.redis.brpoplpush(
            "tasks",           // Source queue
            "tasks:processing", // Processing queue (for recovery)
            30                  // Timeout seconds
        );
        return data ? JSON.parse(data) : null;
    }
    
    async acknowledge(task: Task): Promise<void> {
        await this.redis.lrem("tasks:processing", 1, JSON.stringify(task));
    }
}
 
// PUBLISH-SUBSCRIBE: All consumers receive all events
class EventBroadcaster {
    private readonly subscribers = new Map<string, EventHandler[]>();
    
    subscribe(eventType: string, handler: EventHandler): void {
        const handlers = this.subscribers.get(eventType) || [];
        handlers.push(handler);
        this.subscribers.set(eventType, handlers);
    }
    
    async publish(event: Event): Promise<void> {
        const handlers = this.subscribers.get(event.type) || [];
        
        // All subscribers receive the event
        await Promise.all(
            handlers.map(handler => handler.handle(event))
        );
    }
}
 
// COMPETING CONSUMERS: Consumer groups in Kafka
class KafkaCompetingConsumers {
    async configureConsumerGroup(): Promise<void> {
        const consumer = this.kafka.consumer({
            groupId: "order-processors",  // Multiple instances share this group
        });
        
        await consumer.subscribe({ 
            topic: "orders",
            fromBeginning: false,
        });
        
        // Kafka automatically balances partitions across group members
        // Each message goes to exactly one consumer in the group
        await consumer.run({
            eachMessage: async ({ message }) => {
                await this.processOrder(message);
            },
        });
    }
}
 
// CONTENT-BASED ROUTER: Route based on event content
class ContentBasedRouter {
    private readonly routes: Array<{
        condition: (event: Event) => boolean;
        channel: EventChannel;
    }> = [];
    
    addRoute(
        condition: (event: Event) => boolean,
        channel: EventChannel
    ): void {
        this.routes.push({ condition, channel });
    }
    
    async route(event: Event): Promise<void> {
        for (const route of this.routes) {
            if (route.condition(event)) {
                await route.channel.send(event);
                return;  // First match wins
            }
        }
        // Default channel for unmatched events
        await this.defaultChannel.send(event);
    }
}
 
// Usage: Route orders by priority
const router = new ContentBasedRouter();
router.addRoute(
    (e) => e.payload.priority === "EXPRESS",
    expressOrderChannel
);
router.addRoute(
    (e) => e.payload.priority === "STANDARD",
    standardOrderChannel
);

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

// PATTERN 1: Event ID tracking (deduplication table)
class IdempotentEventHandler {
    constructor(
        private processedEvents: ProcessedEventRepository,
        private businessLogic: BusinessLogic,
    ) {}
    
    async handle(event: Event): Promise<void> {
        // Check if already processed
        const wasProcessed = await this.processedEvents.exists(event.eventId);
        if (wasProcessed) {
            console.log(`Skipping duplicate event: ${event.eventId}`);
            return;
        }
        
        // Process in transaction with deduplication record
        await this.db.transaction(async (tx) => {
            // Record that we're processing this event
            await this.processedEvents.insert(event.eventId, tx);
            
            // Execute business logic
            await this.businessLogic.process(event, tx);
        });
    }
}
 
// PATTERN 2: Natural idempotency via domain design
class NaturallyIdempotentHandler {
    async handleOrderPlaced(event: OrderPlacedEvent): Promise<void> {
        // Check current state rather than tracking event IDs
        const existingReservation = await this.inventoryRepository
            .findReservationByOrderId(event.payload.orderId);
        
        if (existingReservation) {
            // Already processed - idempotent by design
            console.log(`Inventory already reserved for order ${event.payload.orderId}`);
            return;
        }
        
        // First time: create reservation
        await this.inventoryRepository.createReservation({
            orderId: event.payload.orderId,
            items: event.payload.items,
        });
    }
}
 
// PATTERN 3: Idempotent operations (database level)
class IdempotentDatabaseOperations {
    // UPSERT instead of INSERT
    async setCustomerTier(event: CustomerTierChangedEvent): Promise<void> {
        // Runs identical whether first or duplicate
        await this.db.query(`
            INSERT INTO customer_tiers (customer_id, tier, updated_at)
            VALUES ($1, $2, $3)
            ON CONFLICT (customer_id)
            DO UPDATE SET 
                tier = EXCLUDED.tier,
                updated_at = EXCLUDED.updated_at
        `, [event.customerId, event.newTier, event.occurredAt]);
    }
    
    // Conditional update with version check
    async updateBalance(event: PaymentReceivedEvent): Promise<void> {
        const result = await this.db.query(`
            UPDATE accounts 
            SET balance = balance + $1,
                last_transaction_id = $2
            WHERE id = $3 
              AND (last_transaction_id IS NULL OR last_transaction_id < $2)
        `, [event.amount, event.transactionId, event.accountId]);
        
        if (result.rowCount === 0) {
            // Either account doesn't exist, or already processed
            console.log(`Payment ${event.transactionId} already applied or account missing`);
        }
    }
}
 
// PATTERN 4: Idempotency keys for API calls
class ExternalApiIdempotency {
    async processRefund(event: RefundRequestedEvent): Promise<void> {
        // Use event ID as idempotency key for payment provider
        const response = await this.paymentGateway.createRefund({
            amount: event.amount,
            transactionId: event.originalTransactionId,
            idempotencyKey: event.eventId,  // Payment gateway deduplicates
        });
        
        // If duplicate, payment gateway returns existing refund
        // Either way, we get the right result
    }
}

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

interface DeadLetterMessage {
    originalEvent: Event;
    failureReason: string;
    failureStack?: string;
    retryCount: number;
    firstFailedAt: Date;
    lastFailedAt: Date;
    sourceQueue: string;
}
 
class ResilientEventProcessor {
    private readonly maxRetries = 5;
    private readonly baseDelayMs = 1000;
    
    async process(event: Event): Promise<void> {
        let lastError: Error | null = null;
        
        for (let attempt = 1; attempt <= this.maxRetries; attempt++) {
            try {
                await this.handler.handle(event);
                return;  // Success!
            } catch (error) {
                lastError = error as Error;
                
                // Is this a transient or permanent failure?
                if (this.isPermanentFailure(error)) {
                    break;  // Don't retry permanent failures
                }
                
                // Exponential backoff
                const delay = this.baseDelayMs * Math.pow(2, attempt - 1);
                console.log(
                    `Retry ${attempt}/${this.maxRetries} for ${event.eventId}, waiting ${delay}ms`
                );
                await sleep(delay);
            }
        }
        
        // Exhausted retries or permanent failure
        await this.sendToDeadLetter(event, lastError!);
    }
    
    private isPermanentFailure(error: unknown): boolean {
        // Validation errors, business rule violations, etc.
        return error instanceof ValidationError ||
               error instanceof BusinessRuleViolation ||
               error instanceof DeserializationError;
    }
    
    private async sendToDeadLetter(event: Event, error: Error): Promise<void> {
        const dlqMessage: DeadLetterMessage = {
            originalEvent: event,
            failureReason: error.message,
            failureStack: error.stack,
            retryCount: this.maxRetries,
            firstFailedAt: new Date(),
            lastFailedAt: new Date(),
            sourceQueue: this.sourceQueueName,
        };
        
        await this.deadLetterQueue.send(dlqMessage);
        
        // Alert on-call team
        await this.alerting.notify({
            severity: "warning",
            title: `Event sent to DLQ: ${event.eventType}`,
            details: error.message,
            eventId: event.eventId,
        });
    }
}
 
// DLQ monitoring and management
class DeadLetterManager {
    async getDeadLetterStats(): Promise<DLQStats> {
        const messages = await this.dlq.peek(1000);
        
        return {
            totalCount: messages.length,
            byEventType: this.groupBy(messages, m => m.originalEvent.eventType),
            byFailureReason: this.groupBy(messages, m => m.failureReason),
            oldestMessage: messages[0]?.firstFailedAt,
        };
    }
    
    async replay(messageId: string): Promise<void> {
        const dlqMessage = await this.dlq.get(messageId);
        
        // Send back to original queue
        await this.eventBus.publish(
            dlqMessage.originalEvent,
            { topic: dlqMessage.sourceQueue }
        );
        
        // Remove from DLQ
        await this.dlq.delete(messageId);
        
        console.log(`Replayed message ${messageId}`);
    }
    
    async replayAll(filter?: { eventType?: string }): Promise<number> {
        let replayed = 0;
        const messages = await this.dlq.getAll(filter);
        
        for (const message of messages) {
            await this.replay(message.id);
            replayed++;
        }
        
        return replayed;
    }
}

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

// STRATEGY 1: Partition by aggregate ID (Kafka style)
class PartitionedEventPublisher {
    async publish(event: AggregateEvent): Promise<void> {
        await this.kafka.send({
            topic: "order-events",
            messages: [{
                // Key determines partition
                // All events for same order go to same partition
                // Same partition = same consumer = ordering preserved
                key: event.aggregateId,
                value: JSON.stringify(event),
            }],
        });
    }
}
 
// STRATEGY 2: Sequence numbers for out-of-order detection
interface SequencedEvent extends Event {
    sequenceNumber: number;  // Per-aggregate sequence
    aggregateVersion: number;  // Current aggregate version
}
 
class SequenceAwareHandler {
    private readonly lastProcessed = new Map<string, number>();
    
    async handle(event: SequencedEvent): Promise<void> {
        const lastSeq = this.lastProcessed.get(event.aggregateId) ?? 0;
        
        if (event.sequenceNumber <= lastSeq) {
            // Duplicate or old event - skip
            console.log(`Skipping old event: seq ${event.sequenceNumber} <= ${lastSeq}`);
            return;
        }
        
        if (event.sequenceNumber > lastSeq + 1) {
            // Gap detected - events arrived out of order
            console.warn(
                `Gap detected: expected ${lastSeq + 1}, got ${event.sequenceNumber}`
            );
            // Options: wait for missing event, fetch from source, or process anyway
            await this.handleGap(event, lastSeq);
            return;
        }
        
        // Correct sequence - process
        await this.processEvent(event);
        this.lastProcessed.set(event.aggregateId, event.sequenceNumber);
    }
    
    private async handleGap(event: SequencedEvent, lastSeq: number): Promise<void> {
        // Option 1: Buffer and wait
        await this.bufferForLater(event);
        
        // Option 2: Fetch missing events from source
        const missing = await this.eventStore.getEvents(
            event.aggregateId,
            { fromSequence: lastSeq + 1, toSequence: event.sequenceNumber - 1 }
        );
        for (const missedEvent of missing) {
            await this.handle(missedEvent);
        }
        await this.handle(event);  // Retry original event
    }
}
 
// STRATEGY 3: Version-based optimistic handling
class VersionAwareHandler {
    async handleOrderStatusChange(event: OrderStatusChangedEvent): Promise<void> {
        const result = await this.db.query(`
            UPDATE orders 
            SET status = $1, 
                version = $2,
                updated_at = NOW()
            WHERE id = $3 AND version < $4
        `, [
            event.newStatus,
            event.aggregateVersion,
            event.orderId,
            event.aggregateVersion,  // Only update if our version is newer
        ]);
        
        if (result.rowCount === 0) {
            // Either order doesn't exist, or we have stale event
            console.log(
                `Skipping stale status update for order ${event.orderId}`
            );
        }
    }
}