TypeScript's event-driven architecture story has evolved from a Node.js curiosity to a production-grade platform. With the rise of NestJS microservices, native Kafka clients, and TypeScript's type system catching integration errors at build time, it's now a compelling choice for teams that want type safety without sacrificing development velocity.
Architecture Foundation
TypeScript event-driven systems typically run on Node.js with either kafkajs (pure TypeScript) or node-rdkafka (librdkafka bindings). KafkaJS is the default choice — it requires no native dependencies, works everywhere Node.js runs, and provides a clean async/await API.
typescript
1// Event type definitions
2interface BaseEvent {
3 eventId: string;
4 timestamp: string;
5 eventType: string;
6}
7
8interface OrderCreated extends BaseEvent {
9 eventType: "OrderCreated";
10 orderId: string;
11 customerId: string;
12 items: OrderItem[];
13 total: number;
14 currency: string;
15}
16
17interface OrderShipped extends BaseEvent {
18 eventType: "OrderShipped";
19 orderId: string;
20 trackingNumber: string;
21 carrier: string;
22}
23
24interface OrderCancelled extends BaseEvent {
25 eventType: "OrderCancelled";
26 orderId: string;
27 reason: string;
28}
29
30type OrderEvent = OrderCreated | OrderShipped | OrderCancelled;
31
32interface OrderItem {
33 sku: string;
34 quantity: number;
35 unitPrice: number;
36}
37
Discriminated unions with the eventType field let TypeScript narrow types automatically in switch statements — the compiler verifies exhaustive handling.
KafkaJS Producer
A production-ready producer with idempotency, compression, and proper error handling:
typescript
1import { Kafka, Producer, CompressionTypes, Partitioners } from "kafkajs";
2import { randomUUID } from "crypto";
3
4class EventPublisher {
5 private producer: Producer;
6 private connected = false;
7
8 constructor(private kafka: Kafka) {
9 this.producer = kafka.producer({
10 createPartitioner: Partitioners.DefaultPartitioner,
11 idempotent: true,
12 maxInFlightRequests: 5,
13 retry: {
14 retries: 3,
15 initialRetryTime: 100,
16 factor: 2,
17 },
18 });
19 }
20
21 async connect(): Promise<void> {
22 await this.producer.connect();
23 this.connected = true;
24 }
25
26 async publish<T extends BaseEvent>(
27 topic: string,
28 key: string,
29 event: T
30 ): Promise<void> {
31 if (!this.connected) {
32 throw new Error("Producer not connected");
33 }
34
35 await this.producer.send({
36 topic,
37 compression: CompressionTypes.LZ4,
38 messages: [
39 {
40 key,
41 value: JSON.stringify(event),
42 headers: {
43 event_type: event.eventType,
44 event_id: event.eventId,
45 produced_at: new Date().toISOString(),
46 },
47 },
48 ],
49 });
50 }
51
52 async publishBatch<T extends BaseEvent>(
53 topic: string,
54 events: { key: string; event: T }[]
55 ): Promise<void> {
56 await this.producer.send({
57 topic,
58 compression: CompressionTypes.LZ4,
59 messages: events.map(({ key, event }) => ({
60 key,
61 value: JSON.stringify(event),
62 headers: {
63 event_type: event.eventType,
64 event_id: event.eventId,
65 produced_at: new Date().toISOString(),
66 },
67 })),
68 });
69 }
70
71 async disconnect(): Promise<void> {
72 await this.producer.disconnect();
73 this.connected = false;
74 }
75}
76
Consumer with Manual Offset Management
typescript
1import { Consumer, EachMessagePayload } from "kafkajs";
2
3type EventHandler<T> = (event: T) => Promise<void>;
4
5class EventConsumer {
6 private consumer: Consumer;
7 private handlers = new Map<string, EventHandler<unknown>>();
8
9 constructor(
10 private kafka: Kafka,
11 private groupId: string,
12 private topics: string[]
13 ) {
14 this.consumer = kafka.consumer({
15 groupId,
16 sessionTimeout: 30000,
17 heartbeatInterval: 3000,
18 maxWaitTimeInMs: 250,
19 retry: { retries: 5 },
20 });
21 }
22
23 register<T extends BaseEvent>(
24 eventType: string,
25 handler: EventHandler<T>
26 ): void {
27 this.handlers.set(eventType, handler as EventHandler<unknown>);
28 }
29
30 async start(): Promise<void> {
31 await this.consumer.connect();
32 await this.consumer.subscribe({
33 topics: this.topics,
34 fromBeginning: false,
35 });
36
37 await this.consumer.run({
38 autoCommit: false,
39 eachMessage: async (payload: EachMessagePayload) => {
40 const { topic, partition, message } = payload;
41 const eventType = message.headers?.event_type?.toString();
42
43 if (!eventType || !message.value) {
44 console.warn(
45 `Skipping message with missing type at ${topic}:${partition}:${message.offset}`
46 );
47 await this.commit(payload);
48 return;
49 }
50
51 const handler = this.handlers.get(eventType);
52 if (!handler) {
53 console.warn(`No handler for event type: ${eventType}`);
54 await this.commit(payload);
55 return;
56 }
57
58 try {
59 const event = JSON.parse(message.value.toString());
60 await handler(event);
61 await this.commit(payload);
62 } catch (error) {
63 console.error(
64 `Handler failed for ${eventType} at offset ${message.offset}:`,
65 error
66 );
67 await this.sendToDlq(payload, error as Error);
68 await this.commit(payload);
69 }
70 },
71 });
72 }
73
74 private async commit(payload: EachMessagePayload): Promise<void> {
75 await this.consumer.commitOffsets([
76 {
77 topic: payload.topic,
78 partition: payload.partition,
79 offset: (Number(payload.message.offset) + 1).toString(),
80 },
81 ]);
82 }
83
84 private async sendToDlq(
85 payload: EachMessagePayload,
86 error: Error
87 ): Promise<void> {
88 // DLQ publishing implementation
89 }
90
91 async stop(): Promise<void> {
92 await this.consumer.disconnect();
93 }
94}
95
Type-Safe Event Router with Zod Validation
Use Zod schemas for runtime validation that mirrors your TypeScript types:
typescript
1import { z } from "zod";
2
3const OrderCreatedSchema = z.object({
4 eventType: z.literal("OrderCreated"),
5 eventId: z.string().uuid(),
6 orderId: z.string(),
7 customerId: z.string(),
8 items: z.array(
9 z.object({
10 sku: z.string(),
11 quantity: z.number().positive(),
12 unitPrice: z.number().positive(),
13 })
14 ),
15 total: z.number().positive(),
16 currency: z.string().length(3),
17 timestamp: z.string().datetime(),
18});
19
20const OrderShippedSchema = z.object({
21 eventType: z.literal("OrderShipped"),
22 eventId: z.string().uuid(),
23 orderId: z.string(),
24 trackingNumber: z.string(),
25 carrier: z.string(),
26 timestamp: z.string().datetime(),
27});
28
29type ValidatedOrderCreated = z.infer<typeof OrderCreatedSchema>;
30
31class TypedEventRouter {
32 private routes: Map<
33 string,
34 { schema: z.ZodSchema; handler: (event: unknown) => Promise<void> }
35 > = new Map();
36
37 register<T>(
38 eventType: string,
39 schema: z.ZodSchema<T>,
40 handler: (event: T) => Promise<void>
41 ): void {
42 this.routes.set(eventType, {
43 schema,
44 handler: handler as (event: unknown) => Promise<void>,
45 });
46 }
47
48 async dispatch(eventType: string, rawData: unknown): Promise<void> {
49 const route = this.routes.get(eventType);
50 if (!route) {
51 throw new Error(`No route for event type: ${eventType}`);
52 }
53
54 const parsed = route.schema.parse(rawData);
55 await route.handler(parsed);
56 }
57}
58
59// Usage
60const router = new TypedEventRouter();
61router.register("OrderCreated", OrderCreatedSchema, async (event) => {
62 // event is fully typed as ValidatedOrderCreated
63 console.log(`Order ${event.orderId} created for ${event.customerId}`);
64});
65
Need a second opinion on your system design architecture?
I run free 30-minute strategy calls for engineering teams tackling this exact problem.
Book a Free CallTransactional Outbox with Prisma
typescript
1import { PrismaClient } from "@prisma/client";
2import { randomUUID } from "crypto";
3
4const prisma = new PrismaClient();
5
6class OrderService {
7 async createOrder(cmd: CreateOrderCommand): Promise<Order> {
8 const orderId = randomUUID();
9
10 const event: OrderCreated = {
11 eventId: randomUUID(),
12 eventType: "OrderCreated",
13 orderId,
14 customerId: cmd.customerId,
15 items: cmd.items,
16 total: cmd.items.reduce((sum, i) => sum + i.unitPrice * i.quantity, 0),
17 currency: "USD",
18 timestamp: new Date().toISOString(),
19 };
20
21 const [order] = await prisma.$transaction([
22 prisma.order.create({
23 data: {
24 id: orderId,
25 customerId: cmd.customerId,
26 items: cmd.items as any,
27 status: "created",
28 },
29 }),
30 prisma.outboxEvent.create({
31 data: {
32 id: event.eventId,
33 aggregateId: orderId,
34 aggregateType: "Order",
35 eventType: "OrderCreated",
36 payload: event as any,
37 },
38 }),
39 ]);
40
41 return order;
42 }
43}
44
45class OutboxPoller {
46 constructor(
47 private publisher: EventPublisher,
48 private intervalMs = 100
49 ) {}
50
51 async start(): Promise<void> {
52 const poll = async () => {
53 const entries = await prisma.outboxEvent.findMany({
54 where: { publishedAt: null },
55 orderBy: { createdAt: "asc" },
56 take: 100,
57 });
58
59 for (const entry of entries) {
60 try {
61 await this.publisher.publish(
62 `${entry.aggregateType.toLowerCase()}-events`,
63 entry.aggregateId,
64 entry.payload as BaseEvent
65 );
66
67 await prisma.outboxEvent.update({
68 where: { id: entry.id },
69 data: { publishedAt: new Date() },
70 });
71 } catch (error) {
72 console.error(`Failed to publish outbox entry ${entry.id}:`, error);
73 }
74 }
75 };
76
77 setInterval(poll, this.intervalMs);
78 }
79}
80
Concurrency Control
Node.js is single-threaded, so concurrency means managing async operations. Use a semaphore pattern to bound concurrent handler executions:
typescript
1class Semaphore {
2 private queue: (() => void)[] = [];
3 private current = 0;
4
5 constructor(private max: number) {}
6
7 async acquire(): Promise<void> {
8 if (this.current < this.max) {
9 this.current++;
10 return;
11 }
12 return new Promise<void>((resolve) => this.queue.push(resolve));
13 }
14
15 release(): void {
16 this.current--;
17 const next = this.queue.shift();
18 if (next) {
19 this.current++;
20 next();
21 }
22 }
23}
24
25class ConcurrentConsumer {
26 private semaphore: Semaphore;
27
28 constructor(maxConcurrency: number) {
29 this.semaphore = new Semaphore(maxConcurrency);
30 }
31
32 async processMessage(handler: () => Promise<void>): Promise<void> {
33 await this.semaphore.acquire();
34 try {
35 await handler();
36 } finally {
37 this.semaphore.release();
38 }
39 }
40}
41
Observability with OpenTelemetry
typescript
1import { trace, context, propagation, SpanKind } from "@opentelemetry/api";
2
3const tracer = trace.getTracer("event-consumer");
4
5async function instrumentedHandler(
6 message: EachMessagePayload,
7 handler: EventHandler<unknown>
8): Promise<void> {
9 const headers: Record<string, string> = {};
10 for (const [key, value] of Object.entries(message.message.headers || {})) {
11 if (value) headers[key] = value.toString();
12 }
13
14 const parentContext = propagation.extract(context.active(), headers);
15
16 return context.with(parentContext, async () => {
17 const span = tracer.startSpan(
18 `process ${message.topic}`,
19 {
20 kind: SpanKind.CONSUMER,
21 attributes: {
22 "messaging.system": "kafka",
23 "messaging.destination": message.topic,
24 "messaging.kafka.partition": message.partition,
25 "messaging.kafka.offset": Number(message.message.offset),
26 },
27 },
28 parentContext
29 );
30
31 try {
32 const event = JSON.parse(message.message.value!.toString());
33 await handler(event);
34 span.setStatus({ code: 0 });
35 } catch (error) {
36 span.setStatus({ code: 2, message: (error as Error).message });
37 throw error;
38 } finally {
39 span.end();
40 }
41 });
42}
43
Conclusion
TypeScript brings a unique value proposition to event-driven architecture: the type system catches schema mismatches and missing handler cases at build time, while Node.js's event loop model naturally suits the I/O-heavy nature of consuming and producing messages. The combination of KafkaJS for transport, Zod for runtime validation, and discriminated unions for type-safe dispatch creates a development experience where most integration errors surface before code reaches production.
The single-threaded nature of Node.js is both a constraint and a simplification. You don't need to worry about data races or thread synchronization, but you do need to manage concurrency through async patterns and potentially scale across processes for CPU-intensive workloads. For the vast majority of event consumers that perform I/O-bound operations — database writes, HTTP calls, cache updates — a single Node.js process handles thousands of events per second without breaking a sweat.