Server-Sent Events

The SSE channel pushes a notification to a notifiable's live Server-Sent Events stream. It builds on NestJS's native @Sse() support — the channel publishes into an SseHub, and you mount the streaming endpoint yourself with Nest's own @Sse() decorator. No extra transport, no socket server: just the SSE Nest already speaks.

Install

pnpm add @dudousxd/nestjs-notifications-sse

npm install @dudousxd/nestjs-notifications-sse

Register the channel

import { SseChannelModule } from '@dudousxd/nestjs-notifications-sse';

@Module({
  imports: [
    SseChannelModule.forRoot({ event: 'notification' }),
  ],
})
export class AppModule {}

SseChannelModule.forRoot() takes:

The module provides two things: the SseChannel (which the core discovers and pushes into) and the SseHub (an in-memory fan-out you read from in your controller).

The notification side

Annotate the payload method with the @Sse() handle and return a plain object — that's the data delivered to subscribers:

import { type Notifiable, Notification } from '@dudousxd/nestjs-notifications-core';
import { Sse } from '@dudousxd/nestjs-notifications-sse';

@Notification()
export class InvoicePaid {
  constructor(private invoiceId: string) {}

  @Sse()
  toSse({ notifiable }: ChannelContext): Record<string, unknown> {
    return { type: 'invoice.paid', invoiceId: this.invoiceId };
  }
}

The Sse handle also works as a via() token (via() { return [Sse]; }) for explicit routing; implement SseNotification alongside the decorator. If toSse() is absent the channel falls back to toArray(), then to a structural copy of the notification.

Mount the stream — native @Sse()

This package does not mount an endpoint for you. Add a controller method with Nest's native @Sse() and return hub.stream(...). Build the stream key with sseKey() so it matches exactly what the channel publishes to:

import { Controller, Req, Sse, type MessageEvent } from '@nestjs/common';
import type { Observable } from 'rxjs';
import { SseHub, sseKey } from '@dudousxd/nestjs-notifications-sse';

@Controller('notifications')
export class NotificationsController {
  constructor(private readonly hub: SseHub) {}

  @Sse('stream')
  stream(@Req() req: any): Observable<MessageEvent> {
    // build the SAME key the notifiable routes to (tenant-aware)
    return this.hub.stream(sseKey(req.tenantId, String(req.user.id)));
  }
}

The browser opens an EventSource('/notifications/stream'); every notification routed to the SSE channel for that user lands as a MessageEvent on the stream. The hub keeps one subject per key and tears it down when the last subscriber disconnects, so multiple tabs share a stream cleanly.

Or let the library mount it: createNotificationsStreamController

If you'd rather not hand-write that controller, the package ships a factory that builds it for you — createNotificationsStreamController. You supply how to resolve the route (and optionally the tenant) from the request; it mounts a native @Sse() endpoint that subscribes to the SseHub under the same key the channel publishes to, and adds a keep-alive heartbeat:

import { Module } from '@nestjs/common';
import { createNotificationsStreamController } from '@dudousxd/nestjs-notifications-sse';
import { AuthGuard } from './auth.guard';

const NotificationsStreamController = createNotificationsStreamController({
  resolveRoute: (req) => String(req.user.id),          // must match routeNotificationFor('sse')
  resolveTenant: (req) => req.tenantId,                // optional; match the send's tenant scope
  guards: [AuthGuard],                                 // the stream is per-user — protect it
});

@Module({ controllers: [NotificationsStreamController] })
export class InboxModule {}

That mounts GET notifications/stream. NotificationsStreamControllerOptions:

Under the hood it does exactly what the hand-written controller does — builds sseKey(tenant, route) and subscribes to this.hub.stream(key) — plus merges in the heartbeat. Requires SseChannelModule (which provides SseHub) in scope, and pairs with a cross-pod backplane so a publish on any node reaches connections held by another.

Routing and tenant-awareness

The per-user route comes from routeNotificationFor('sse') — typically the user id. The channel combines it with the delivery's tenant to form the publish key, so a user's stream is isolated per tenant:

export class User implements Notifiable {
  constructor(public id: string) {}

  routeNotificationFor(channel: string) {
    if (channel !== 'sse') return undefined;
    return this.id;
  }
}

sseKey(tenant, routeValue) prefixes the route with ${tenant}: when a tenant is present. As long as your controller builds the key with the same sseKey(req.tenantId, ...), a tenant-scoped send (via forTenant(id) or a @Tenant() property) reaches exactly the right stream.

Scaling across instances

By default the SseHub fans a notification out in-process only — a stream connected to pod A never sees a notification published on pod B. Pass a backplane and every publish fans out over it to every pod, so a user's stream stays live no matter which pod is running the send:

import Redis from 'ioredis';
import { SseChannelModule, redisSseBackplane } from '@dudousxd/nestjs-notifications-sse';

@Module({
  imports: [
    SseChannelModule.forRoot({
      event: 'notification',
      backplane: redisSseBackplane(() => new Redis(process.env.REDIS_URL)),
    }),
  ],
})
export class AppModule {}

redisSseBackplane(createClient, options?) builds a RedisSseBackplane by calling createClient() twice — one publisher, one subscriber. That's not incidental: a client that has entered subscribe mode rejects regular commands (ioredis fails with "Connection in subscriber mode, only subscriber commands may be used"), so sharing one connection for both roles is a footgun the factory exists to prevent. The package stays BYO — it doesn't depend on ioredis itself, so you control the client version/config; createClient just needs to return something satisfying the RedisPubSubClient interface (publish, subscribe, on('message', ...), optional quit).

Prefer building the two clients yourself (e.g. from a DI-injected Redis config)? Construct RedisSseBackplane directly instead of the factory:

import { RedisSseBackplane } from '@dudousxd/nestjs-notifications-sse';

new RedisSseBackplane({ publisher: new Redis(url), subscriber: new Redis(url) });

Cross-device read sync

The same stream also carries read events. Bind SseReadSyncPublisher under the database channel's READ_SYNC_PUBLISHER token and, when a user marks a notification read, their other open streams get an event: read frame — so the unread badge clears everywhere without a refetch. Full setup in the Real-time & in-app guide.