Lately, I’ve been thinking a lot about how to build systems that are both resilient and easy to maintain. The challenge of handling numerous services communicating in real-time led me to explore event-driven architectures. Today, I want to share a practical approach to implementing such a system using TypeScript, NestJS, and Redis Streams, ensuring everything remains type-safe and scalable.

Why does type safety matter in event-driven systems? Imagine trying to debug why an event handler fails because of a mismatched property type. By defining strict event schemas, we prevent these issues early. Here’s how you can define a basic event structure:

interface BaseEvent {
  eventId: string;
  eventType: string;
  timestamp: Date;
  payload: Record<string, any>;
}

class OrderCreatedEvent implements BaseEvent {
  eventId: string;
  eventType = 'order.created';
  timestamp: Date;
  payload: {
    orderId: string;
    customerId: string;
    amount: number;
  };
}

Setting up Redis Streams in NestJS is straightforward. Redis Streams offer persistence, consumer groups, and message ordering—features essential for robust event processing. First, install the necessary packages:

npm install redis @nestjs/common @nestjs/core

Then, create a service to handle publishing events:

import { Injectable } from '@nestjs/common';
import { createClient } from 'redis';

@Injectable()
export class EventPublisher {
  private client;

  constructor() {
    this.client = createClient({ url: 'redis://localhost:6379' });
    this.client.connect();
  }

  async publish(stream: string, event: BaseEvent) {
    await this.client.xAdd(stream, '*', { ...event });
  }
}

How do we ensure that events are processed correctly even when services restart? Consumer groups in Redis Streams allow multiple services to read from the same stream without missing messages. Here’s a basic consumer setup:

import { Injectable } from '@nestjs/common';
import { createClient } from 'redis';

@Injectable()
export class EventConsumer {
  private client;

  constructor() {
    this.client = createClient({ url: 'redis://localhost:6379' });
    this.client.connect();
  }

  async consume(stream: string, group: string, consumer: string) {
    while (true) {
      const messages = await this.client.xReadGroup(
        group, consumer, [{ key: stream, id: '>' }], { COUNT: 10 }
      );
      // Process messages here
    }
  }
}

Handling errors gracefully is critical. What happens if an event fails processing? Implementing a dead-letter queue ensures failed events are set aside for later review without blocking the stream:

async handleEvent(message) {
  try {
    await processMessage(message);
    await this.client.xAck('mystream', 'mygroup', message.id);
  } catch (error) {
    await this.client.xAdd('dead-letter-queue', '*', message);
  }
}

Adding type safety to event handlers reduces runtime errors. Using TypeScript decorators, we can ensure each handler receives the correct event type:

function EventHandler(eventType: string) {
  return function (target: any, propertyKey: string, descriptor: PropertyDescriptor) {
    // Register handler logic here
  };
}

class OrderService {
  @EventHandler('order.created')
  handleOrderCreated(event: OrderCreatedEvent) {
    // Safe to use event.payload.amount as number
  }
}

Monitoring event flows helps identify bottlenecks. Integrating with tools like Prometheus or emitting logs for each processed event provides visibility into system health.

In my experience, starting with a well-defined event schema and leveraging Redis Streams’ built-in features creates a foundation that scales effortlessly. Have you considered how event ordering might impact your business logic?

I encourage you to try building a small project using this approach. Feel free to share your thoughts or questions in the comments below—I’d love to hear about your experiences and answer any questions you might have. If you found this useful, please like and share it with others who might benefit.