I’ve been thinking about how modern applications need to handle events reliably while maintaining type safety across distributed components. This led me to explore combining NestJS, Redis Streams, and Prisma - a stack that ensures both robustness and developer productivity. When services communicate through events rather than direct calls, we gain scalability and fault tolerance, but how do we prevent type errors from creeping in? That’s the challenge I’ll address here.

Let’s start by setting up our foundation. We’ll create a new NestJS project and install essential packages:

npm i @nestjs/cli
nest new event-app
cd event-app
npm install @prisma/client prisma redis ioredis zod
npx prisma init

The project structure organizes concerns logically:

src/
├── events/
│   ├── schemas/
│   ├── bus.service.ts
│   └── store.service.ts
├── users/
│   ├── user.events.ts
│   └── user.service.ts
└── app.module.ts

For type safety, we define events using Zod schemas. Consider this user creation event:

// src/events/schemas/user.schema.ts
import { z } from 'zod';

export const UserCreatedSchema = z.object({
  id: z.string().uuid(),
  type: z.literal('UserCreated'),
  timestamp: z.date(),
  data: z.object({
    email: z.string().email(),
    name: z.string(),
    role: z.enum(['admin','user'])
  })
});
export type UserCreatedEvent = z.infer<typeof UserCreatedSchema>;

Validation happens at both compile time and runtime. What happens if someone tries to publish an invalid event? Our system needs to catch that early. Here’s how we implement the Redis event bus:

// src/events/bus.service.ts
import { Injectable } from '@nestjs/common';
import Redis from 'ioredis';

@Injectable()
export class EventBusService {
  private readonly redis = new Redis();

  async publish(stream: string, event: object) {
    await this.redis.xadd(stream, '*', 'event', JSON.stringify(event));
  }

  async consume(stream: string, group: string, consumer: string) {
    return this.redis.xreadgroup(
      'GROUP', group, consumer,
      'COUNT', '1',
      'STREAMS', stream, '>'
    );
  }
}

Notice we’re using Redis consumer groups - they enable reliable message processing across multiple instances. But how do we ensure failed events get retried properly? We implement acknowledgments and dead-letter queues:

async handleEvent(stream: string, event: any) {
  try {
    await this.processEvent(event);
    await this.redis.xack(stream, group, eventId);
  } catch (error) {
    await this.redis.xadd('dead-letter-queue', '*', event);
  }
}

For event sourcing with Prisma, we design our schema to store every state change:

// prisma/schema.prisma
model EventStore {
  id        String   @id @default(uuid())
  type      String
  data      Json
  timestamp DateTime @default(now())
  stream    String
}

The event handler becomes straightforward with type guards:

// src/users/user.events.ts
async handleUserCreated(event: unknown) {
  const parsed = UserCreatedSchema.safeParse(event);
  if (!parsed.success) {
    throw new Error('Invalid event format');
  }
  
  const userData = parsed.data;
  await this.prisma.user.create({ data: userData });
}

When implementing producers, we leverage NestJS services with type-safe events:

// src/users/user.service.ts
@Injectable()
export class UserService {
  constructor(private eventBus: EventBusService) {}

  async createUser(dto: CreateUserDto) {
    const user = await this.prisma.user.create({ data: dto });
    const event: UserCreatedEvent = {
      id: uuid(),
      type: 'UserCreated',
      timestamp: new Date(),
      data: user
    };
    await this.eventBus.publish('users-stream', event);
    return user;
  }
}

Monitoring becomes crucial in production. We use Redis commands like XLEN and XINFO to track stream health:

> XLEN users-stream
(integer) 42
> XINFO GROUPS users-stream
1) 1) name
   2) "users-group"

For testing, we verify both event publishing and handling:

it('should publish UserCreated event', async () => {
  await userService.createUser(testUser);
  const events = await eventBus.getStreamEvents('users-stream');
  expect(events).toHaveLength(1);
  expect(events[0].type).toBe('UserCreated');
});

Common pitfalls include forgetting idempotency in handlers and misconfiguring consumer groups. Always set MAXLEN on streams to prevent memory issues and implement proper error logging.

I’ve found this combination provides excellent type safety while maintaining Redis’ performance benefits. The stack works well for microservices needing ordered, persistent event processing. Have you considered how type-safe events could prevent bugs in your current system?

If you’re building distributed systems, this approach offers concrete advantages. Try implementing a small event flow and measure the type safety improvements. What challenges have you faced with event-driven architectures? Share your experiences below - I’d love to hear what works in your projects. If this helped you, please like and share with others who might benefit!