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

/**
 * Observer Pattern Implementation
 * 
 * The subject directly manages its observers.
 * Simpler than full pub-sub but more coupled.
 */
 
// Generic observer interface
interface Observer<T> {
    update(data: T): void;
}
 
// Generic subject (observable)
class Subject<T> {
    private observers: Set<Observer<T>> = new Set();
    
    attach(observer: Observer<T>): void {
        this.observers.add(observer);
    }
    
    detach(observer: Observer<T>): void {
        this.observers.delete(observer);
    }
    
    protected notify(data: T): void {
        for (const observer of this.observers) {
            observer.update(data);
        }
    }
}
 
// Concrete subject: Stock price tracker
interface PriceUpdate {
    symbol: string;
    price: number;
    timestamp: Date;
}
 
class StockTicker extends Subject<PriceUpdate> {
    private prices: Map<string, number> = new Map();
    
    updatePrice(symbol: string, price: number): void {
        this.prices.set(symbol, price);
        
        // Notify all observers of the price change
        this.notify({
            symbol,
            price,
            timestamp: new Date()
        });
    }
    
    getPrice(symbol: string): number | undefined {
        return this.prices.get(symbol);
    }
}
 
// Concrete observers
class PriceDisplay implements Observer<PriceUpdate> {
    update(data: PriceUpdate): void {
        console.log(`[DISPLAY] ${data.symbol}: $${data.price.toFixed(2)}`);
    }
}
 
class PriceLogger implements Observer<PriceUpdate> {
    private log: PriceUpdate[] = [];
    
    update(data: PriceUpdate): void {
        this.log.push(data);
        console.log(`[LOG] Recorded price update #${this.log.length}`);
    }
}
 
class PriceAlertService implements Observer<PriceUpdate> {
    constructor(private readonly thresholds: Map<string, number>) {}
    
    update(data: PriceUpdate): void {
        const threshold = this.thresholds.get(data.symbol);
        if (threshold && data.price < threshold) {
            console.log(`[ALERT] ${data.symbol} dropped below $${threshold}!`);
        }
    }
}
 
// Usage
const ticker = new StockTicker();
 
// Attach observers
const display = new PriceDisplay();
const logger = new PriceLogger();
const alerts = new PriceAlertService(new Map([['AAPL', 150]]));
 
ticker.attach(display);
ticker.attach(logger);
ticker.attach(alerts);
 
// Price updates notify all observers
ticker.updatePrice('AAPL', 155);  // Display, Logger notified
ticker.updatePrice('AAPL', 145);  // Display, Logger, and Alert notified!
 
// Detach when done
ticker.detach(display);

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
176
177
178
179
180
181

/**
 * Production-Ready In-Memory Event Bus
 * 
 * Features:
 * - Type-safe event routing
 * - Async handler execution
 * - Error isolation
 * - Middleware support
 * - Graceful shutdown
 */
 
type EventHandler<T = unknown> = (event: T) => Promise<void> | void;
                            type Middleware = (event: unknown, next: () => Promise<void>) => Promise<void>;
                            type Unsubscribe = () => void;
 
                            interface DomainEvent {
                            eventType: string;
                            eventId: string;
                            occurredAt: Date;
                            [key: string]: unknown;
                }
 
class InMemoryEventBus {
                    private handlers = new Map<string, Set<EventHandler>>();
                    private wildcardHandlers = new Set<EventHandler>();
                    private middlewares: Middleware[] = [];
                    private isShuttingDown = false;
                    private pendingHandlers = 0;
 
                    /**
                     * Register middleware to run on every event.
                     */
                    use(middleware: Middleware): this {
                        this.middlewares.push(middleware);
                        return this;
                    }
 
                    /**
                     * Subscribe to a specific event type.
                     */
                    on<T extends DomainEvent>(eventType: string, handler: EventHandler<T>): Unsubscribe {
                        if (!this.handlers.has(eventType)) {
                            this.handlers.set(eventType, new Set());
                        }
                        this.handlers.get(eventType)!.add(handler as EventHandler);
 
                        return () => {
                            this.handlers.get(eventType)?.delete(handler as EventHandler);
                        };
                    }
 
                    /**
                     * Subscribe to all events (useful for logging, metrics).
                     */
                    onAny(handler: EventHandler<DomainEvent>): Unsubscribe {
                        this.wildcardHandlers.add(handler);
                        return () => this.wildcardHandlers.delete(handler);
                    }
 
                    /**
                     * Publish an event to all interested handlers.
                     */
                    async publish<T extends DomainEvent>(event: T): Promise<void> {
                        if (this.isShuttingDown) {
                            throw new Error('Event bus is shutting down');
                        }
 
                        const typeHandlers = this.handlers.get(event.eventType) || new Set();
                        const allHandlers = [...typeHandlers, ...this.wildcardHandlers];
 
                        if (allHandlers.length === 0) {
                            return;  // No handlers, nothing to do
                        }
 
                        // Build middleware chain
                        const executeHandlers = async () => {
                            this.pendingHandlers += allHandlers.length;
 
                            const results = await Promise.allSettled(
                                allHandlers.map(async handler => {
                                    try {
                                        await handler(event);
                                    } finally {
                                        this.pendingHandlers--;
                                    }
                                })
                            );
 
                            // Log failures but don't throw (isolation)
                            for (const result of results) {
                                if (result.status === 'rejected') {
                                    console.error('Handler failed:', result.reason);
                                }
                            }
                        };
 
                        // Execute through middleware chain
                        await this.executeWithMiddleware(event, executeHandlers);
                    }
 
                    private async executeWithMiddleware(
                        event: unknown,
                        finalHandler: () => Promise<void>
                    ): Promise<void> {
                        let index = 0;
 
                        const next = async (): Promise<void> => {
                            if (index < this.middlewares.length) {
                                const middleware = this.middlewares[index++];
                                await middleware(event, next);
                            } else {
                                await finalHandler();
                            }
                        };
 
                        await next();
                    }
 
                    /**
                     * Gracefully shutdown, waiting for pending handlers.
                     */
                    async shutdown(timeoutMs = 5000): Promise<void> {
                        this.isShuttingDown = true;
 
                        const start = Date.now();
                        while (this.pendingHandlers > 0) {
                            if (Date.now() - start > timeoutMs) {
                                console.warn(`Shutdown timeout with ${this.pendingHandlers} pending handlers`);
                break;
            }
            await new Promise(resolve => setTimeout(resolve, 100));
        }
        
        this.handlers.clear();
        this.wildcardHandlers.clear();
    }
}
 
// ========== Usage Example ==========
 
const eventBus = new InMemoryEventBus();
 
// Add logging middleware
eventBus.use(async (event, next) => {
    const e = event as DomainEvent;
    console.log(`[EVENT] ${e.eventType} - ${e.eventId}`);
    const start = Date.now();
    await next();
    console.log(`[EVENT] ${e.eventType} completed in ${Date.now() - start}ms`);
});
 
// Add metrics middleware
eventBus.use(async (event, next) => {
    const e = event as DomainEvent;
    metrics.increment(`events.${e.eventType}.published`);
    await next();
});
 
// Subscribe to specific events
const unsubscribe = eventBus.on<OrderPlacedEvent>('OrderPlaced', async (event) => {
    console.log(`Processing order ${event.orderId}`);
    await sendConfirmationEmail(event);
});
 
// Subscribe to all events (for audit log)
eventBus.onAny(async (event) => {
    await auditLog.record(event);
});
 
// Publish events
await eventBus.publish({
    eventType: 'OrderPlaced',
    eventId: 'evt-123',
    occurredAt: new Date(),
    orderId: 'order-456',
    customerId: 'cust-789'
});
 
// Cleanup
unsubscribe();
await eventBus.shutdown();

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

/**
 * The Dual Write Problem
 * 
 * Publishing to a message broker and saving to database
 * are two separate operations that can fail independently.
 */
 
// ❌ PROBLEM: Dual writes can lead to inconsistency
async function placeOrderDangerous(command: PlaceOrderCommand): Promise<Order> {
    const order = new Order(command);
    
    // Step 1: Save to database
    await orderRepository.save(order);  // Succeeds
    
    // Step 2: Publish to message broker
    await messageBroker.publish(orderPlacedEvent);  // FAILS!
    
    // Result: Order is saved but event was never published
    // Downstream systems never learn about the order
    // Database and event stream are now INCONSISTENT
    
    return order;
}
 
// Also problematic in the reverse order:
async function placeOrderStillDangerous(command: PlaceOrderCommand): Promise<Order> {
    const order = new Order(command);
    
    // Step 1: Publish first
    await messageBroker.publish(orderPlacedEvent);  // Succeeds
    
    // Step 2: Save to database
    await orderRepository.save(order);  // FAILS!
    
    // Result: Event published but order never saved
    // Downstream systems process a non-existent order
    // Even worse inconsistency!
    
    return 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
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

/**
 * Transactional Outbox Pattern
 * 
 * Events are stored in the database as part of the same transaction.
 * A separate publisher process reads and publishes them.
 */
 
// Outbox table schema (SQL)
/*
CREATE TABLE outbox (
    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 DEFAULT NOW(),
    published_at TIMESTAMP NULL
);
 
CREATE INDEX idx_outbox_unpublished ON outbox (created_at) 
    WHERE published_at IS NULL;
*/
 
interface OutboxEntry {
    id: string;
    aggregateType: string;
    aggregateId: string;
    eventType: string;
    payload: object;
    createdAt: Date;
    publishedAt: Date | null;
}
 
class OrderService {
    constructor(
        private readonly db: Database,
        private readonly orderRepository: IOrderRepository
    ) {}
    
    async placeOrder(command: PlaceOrderCommand): Promise<Order> {
        // Single transaction for both operations
        return this.db.transaction(async (tx) => {
            // Create and save order
            const order = new Order(command);
            await this.orderRepository.save(order, tx);
            
            // Create event for outbox (same transaction!)
            const event = {
                eventType: 'OrderPlaced',
                eventId: generateUUID(),
                occurredAt: new Date(),
                aggregateId: order.id,
                payload: {
                    orderId: order.id,
                    customerId: order.customerId,
                    items: order.items,
                    totalAmount: order.totalAmount
                }
            };
            
            // Write to outbox table (same transaction!)
            await tx.query(`
                INSERT INTO outbox (id, aggregate_type, aggregate_id, event_type, payload)
                VALUES ($1, $2, $3, $4, $5)
            `, [
                event.eventId,
                'Order',
                order.id,
                event.eventType,
                JSON.stringify(event.payload)
            ]);
            
            // Transaction commits: order AND outbox entry are persisted atomically
            return order;
        });
    }
}
 
/**
 * Outbox Publisher: Background process that reads outbox and publishes
 */
class OutboxPublisher {
    private isRunning = false;
    
    constructor(
        private readonly db: Database,
        private readonly messageBroker: IMessageBroker
    ) {}
    
    async start(): Promise<void> {
        this.isRunning = true;
        
        while (this.isRunning) {
            await this.publishPendingEvents();
            await this.sleep(100);  // Poll interval
        }
    }
    
    async stop(): Promise<void> {
        this.isRunning = false;
    }
    
    private async publishPendingEvents(): Promise<void> {
        // Fetch unpublished events
        const entries = await this.db.query<OutboxEntry>(`
            SELECT * FROM outbox 
            WHERE published_at IS NULL 
            ORDER BY created_at 
            LIMIT 100
            FOR UPDATE SKIP LOCKED  -- Allow concurrent publishers
        `);
        
        for (const entry of entries) {
            try {
                // Publish to message broker
                await this.messageBroker.publish({
                    eventType: entry.eventType,
                    eventId: entry.id,
                    aggregateId: entry.aggregateId,
                    payload: entry.payload
                });
                
                // Mark as published
                await this.db.query(`
                    UPDATE outbox SET published_at = NOW() WHERE id = $1
                `, [entry.id]);
                
            } catch (error) {
                console.error(`Failed to publish ${entry.id}:`, error);
                // Will retry on next poll
            }
        }
    }
    
    private sleep(ms: number): Promise<void> {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}
 
/**
 * Cleanup: Periodically delete old published events
 */
async function cleanupOutbox(db: Database): Promise<void> {
    await db.query(`
        DELETE FROM outbox 
        WHERE published_at IS NOT NULL 
        AND published_at < NOW() - INTERVAL '7 days'
    `);
}

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

/**
 * Redis Pub/Sub Implementation
 * 
 * Best for: Real-time notifications, ephemeral events
 * NOT for: Durable messaging, guaranteed delivery
 */
 
import Redis from 'ioredis';
 
class RedisPubSubEventBus {
    private publisher: Redis;
    private subscriber: Redis;
    private handlers = new Map<string, Set<(event: unknown) => void>>();
    
    constructor(redisUrl: string) {
        // Need separate connections for pub and sub
        this.publisher = new Redis(redisUrl);
        this.subscriber = new Redis(redisUrl);
        
        // Handle incoming messages
        this.subscriber.on('message', (channel, message) => {
            const handlers = this.handlers.get(channel);
            if (handlers) {
                const event = JSON.parse(message);
                handlers.forEach(handler => handler(event));
            }
        });
    }
    
    async publish<T extends { eventType: string }>(event: T): Promise<void> {
        const channel = event.eventType;
        const message = JSON.stringify(event);
        
        await this.publisher.publish(channel, message);
    }
    
    async subscribe(
        eventType: string,
        handler: (event: unknown) => void
    ): Promise<() => void> {
        if (!this.handlers.has(eventType)) {
            this.handlers.set(eventType, new Set());
            await this.subscriber.subscribe(eventType);
        }
        
        this.handlers.get(eventType)!.add(handler);
        
        return () => {
            this.handlers.get(eventType)?.delete(handler);
        };
    }
    
    async shutdown(): Promise<void> {
        await this.subscriber.quit();
        await this.publisher.quit();
    }
}
 
// Usage
const eventBus = new RedisPubSubEventBus('redis://localhost:6379');
 
await eventBus.subscribe('OrderPlaced', async (event) => {
    console.log('Order placed:', event);
});
 
await eventBus.publish({
    eventType: 'OrderPlaced',
    orderId: '123',
    customerId: '456'
});

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

/**
 * RabbitMQ Implementation
 * 
 * Best for: Work queues, durable messaging, complex routing
 * Features: Message acknowledgment, dead letter queues, TTL
 */
 
import amqp from 'amqplib';
 
class RabbitMQEventBus {
    private connection: amqp.Connection | null = null;
    private channel: amqp.Channel | null = null;
    private readonly exchangeName = 'domain-events';
    
    async connect(url: string): Promise<void> {
        this.connection = await amqp.connect(url);
        this.channel = await this.connection.createChannel();
        
        // Create topic exchange for flexible routing
        await this.channel.assertExchange(this.exchangeName, 'topic', {
            durable: true  // Survives broker restart
        });
    }
    
    async publish<T extends { eventType: string; eventId: string }>(
        event: T
    ): Promise<void> {
        const routingKey = event.eventType;
        const message = Buffer.from(JSON.stringify(event));
        
        this.channel!.publish(this.exchangeName, routingKey, message, {
            persistent: true,      // Survives broker restart
            contentType: 'application/json',
            messageId: event.eventId,
            timestamp: Date.now()
        });
    }
    
    async subscribe(
        eventPattern: string,  // e.g., 'Order.*' or 'Order.Placed'
        queueName: string,
        handler: (event: unknown) => Promise<void>
    ): Promise<void> {
        // Create durable queue
        await this.channel!.assertQueue(queueName, {
            durable: true,
            deadLetterExchange: 'dlx'  // Failed messages go here
        });
        
        // Bind queue to exchange with routing pattern
        await this.channel!.bindQueue(queueName, this.exchangeName, eventPattern);
        
        // Consume messages
        await this.channel!.consume(queueName, async (msg) => {
            if (!msg) return;
            
            try {
                const event = JSON.parse(msg.content.toString());
                await handler(event);
                
                // Acknowledge successful processing
                this.channel!.ack(msg);
            } catch (error) {
                console.error('Handler failed:', error);
                
                // Negative ack - requeue for retry
                // After max retries, goes to dead letter queue
                this.channel!.nack(msg, false, false);
            }
        });
    }
    
    async shutdown(): Promise<void> {
        await this.channel?.close();
        await this.connection?.close();
    }
}
 
// Usage
const eventBus = new RabbitMQEventBus();
await eventBus.connect('amqp://localhost');
 
// Subscribe with pattern matching
await eventBus.subscribe('Order.*', 'order-notifications', async (event) => {
    console.log('Order event:', event);
});
 
// This matches 'Order.*' pattern
await eventBus.publish({
    eventType: 'Order.Placed',
    eventId: 'evt-123',
    orderId: '456'
});

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

/**
 * Apache Kafka Implementation
 * 
 * Best for: Event streaming, high throughput, event sourcing
 * Features: Persistent log, replay capability, partitioning
 */
 
import { Kafka, Producer, Consumer, EachMessagePayload } from 'kafkajs';
 
class KafkaEventBus {
    private kafka: Kafka;
    private producer: Producer;
    private consumers: Map<string, Consumer> = new Map();
    
    constructor(brokers: string[]) {
        this.kafka = new Kafka({
            clientId: 'my-app',
            brokers
        });
        this.producer = this.kafka.producer();
    }
    
    async connect(): Promise<void> {
        await this.producer.connect();
    }
    
    async publish<T extends { eventType: string; aggregateId: string }>(
        event: T,
        topic = 'domain-events'
    ): Promise<void> {
        await this.producer.send({
            topic,
            messages: [{
                // Partition by aggregateId for ordering guarantees
                key: event.aggregateId,
                value: JSON.stringify(event),
                headers: {
                    'event-type': event.eventType
                }
            }]
        });
    }
    
    async subscribe(
        topic: string,
        groupId: string,
        handler: (event: unknown) => Promise<void>
    ): Promise<void> {
        const consumer = this.kafka.consumer({ groupId });
        await consumer.connect();
        await consumer.subscribe({ topic, fromBeginning: false });
        
        await consumer.run({
            eachMessage: async ({ message }: EachMessagePayload) => {
                const event = JSON.parse(message.value!.toString());
                
                try {
                    await handler(event);
                    // Kafka auto-commits offset after successful processing
                } catch (error) {
                    console.error('Handler failed:', error);
                    // In production, implement retry logic or dead letter
                    throw error;  // Causes reprocessing
                }
            }
        });
        
        this.consumers.set(groupId, consumer);
    }
    
    async shutdown(): Promise<void> {
        for (const consumer of this.consumers.values()) {
            await consumer.disconnect();
        }
        await this.producer.disconnect();
    }
}
 
// Usage
const eventBus = new KafkaEventBus(['localhost:9092']);
await eventBus.connect();
 
// Multiple instances share work via consumer group
await eventBus.subscribe('domain-events', 'order-processors', async (event) => {
    console.log('Processing event:', event);
});
 
// Events partitioned by aggregateId for ordering
await eventBus.publish({
    eventType: 'OrderPlaced',
    aggregateId: 'order-123',
    customerId: 'cust-456'
});

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

/**
 * AWS SNS (topics) + SQS (queues) Implementation
 * 
 * SNS: Fan-out to multiple subscribers
 * SQS: Durable queues with at-least-once delivery
 */
 
import { SNSClient, PublishCommand } from '@aws-sdk/client-sns';
import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from '@aws-sdk/client-sqs';
 
class AWSSNSSQSEventBus {
    private sns: SNSClient;
    private sqs: SQSClient;
    
    constructor(region: string) {
        this.sns = new SNSClient({ region });
        this.sqs = new SQSClient({ region });
    }
    
    async publish<T extends { eventType: string }>(
        event: T,
        topicArn: string
    ): Promise<void> {
        await this.sns.send(new PublishCommand({
            TopicArn: topicArn,
            Message: JSON.stringify(event),
            MessageAttributes: {
                eventType: {
                    DataType: 'String',
                    StringValue: event.eventType
                }
            }
        }));
    }
    
    async consume(
        queueUrl: string,
        handler: (event: unknown) => Promise<void>
    ): Promise<void> {
        while (true) {
            const response = await this.sqs.send(new ReceiveMessageCommand({
                QueueUrl: queueUrl,
                WaitTimeSeconds: 20,  // Long polling
                MaxNumberOfMessages: 10
            }));
            
            for (const message of response.Messages || []) {
                try {
                    // SNS wraps the message
                    const snsMessage = JSON.parse(message.Body!);
                    const event = JSON.parse(snsMessage.Message);
                    
                    await handler(event);
                    
                    // Delete after successful processing
                    await this.sqs.send(new DeleteMessageCommand({
                        QueueUrl: queueUrl,
                        ReceiptHandle: message.ReceiptHandle
                    }));
                } catch (error) {
                    console.error('Handler failed:', error);
                    // Message becomes visible again after visibility timeout
                    // After max retries, goes to dead letter queue
                }
            }
        }
    }
}

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

/**
 * Google Cloud Pub/Sub Implementation
 * 
 * Fully managed, globally distributed, at-least-once delivery
 */
 
import { PubSub, Subscription } from '@google-cloud/pubsub';
 
class GCPPubSubEventBus {
    private pubsub: PubSub;
    
    constructor(projectId: string) {
        this.pubsub = new PubSub({ projectId });
    }
    
    async publish<T extends { eventType: string }>(
        event: T,
        topicName: string
    ): Promise<void> {
        const topic = this.pubsub.topic(topicName);
        
        await topic.publishMessage({
            json: event,
            attributes: {
                eventType: event.eventType
            }
        });
    }
    
    async subscribe(
        subscriptionName: string,
        handler: (event: unknown) => Promise<void>
    ): Promise<void> {
        const subscription = this.pubsub.subscription(subscriptionName);
        
        subscription.on('message', async (message) => {
            try {
                const event = JSON.parse(message.data.toString());
                await handler(event);
                message.ack();  // Acknowledge success
            } catch (error) {
                console.error('Handler failed:', error);
                message.nack();  // Retry later
            }
        });
        
        subscription.on('error', (error) => {
            console.error('Subscription error:', error);
        });
    }
}