Notifications pipeline

The shape for a notifications system that respects user preferences, rate-limits per channel, isolates flaky delivery providers from the request path, and runs on a worker fleet — so an SMS provider outage doesn't slow down API responses.

Shape

• Producer. API pods emit notification events; return immediately without waiting for delivery.
• Queue. Rotif (Redis-backed) carries events to workers, decoupling the request path from delivery.
• Worker. A separate pod consumes events, resolves channel targets per recipient, persists the notification, renders the template, calls the channel.
• Rate-limited. Per-user / per-channel limits prevent runaway loops from spamming users.
• Preferences. Per-user opt-in/out per channel.
• Templates. One payload renders to N channels with per-channel transformations.

Architecture

AppModule — producer (API pod)

import { Module } from '@omnitron-dev/titan';
import { ConfigModule } from '@omnitron-dev/titan/module/config';
import { LoggerModule } from '@omnitron-dev/titan/module/logger';

import { TitanRedisModule } from '@omnitron-dev/titan-redis';
import { EventsModule } from '@omnitron-dev/titan-events';
import { NotificationsModule } from '@omnitron-dev/titan-notifications';

@Module({
  imports: [
    ConfigModule.forRoot({/* … */}),
    LoggerModule.forRoot({ level: 'info' }),

    TitanRedisModule.forRoot({
      config: { url: process.env.REDIS_URL, db: 0 },
    }),

    EventsModule.forRoot({
      wildcard:     true,
      maxListeners: 100,
      history:      { enabled: false },
    }),

    NotificationsModule.forRoot({
      redis: process.env.REDIS_URL,

      // Optional: route through Rotif for cross-pod delivery
      transport: {
        rotif: { /* rotif transport options */ },
      },

      defaultChannels: ['inApp', 'email'],   // channel ids are camelCase: 'inApp', not 'inapp'
      enableInApp:     true,
      enableWebhook:   true,

      templates: {
        enabled:      true,
        cacheEnabled: true,
        cacheTTL:     60_000,
      },

      rateLimiterConfig: {
        defaultLimits: { perMinute: 10, perHour: 100, perDay: 1_000 },
        channelLimits: { sms: { perMinute: 1, perHour: 10 } },
        enableBurstDetection: true,
      },

      preferenceStoreConfig: {
        // Each channel maps to an object, not a bare boolean.
        // Channel ids are camelCase: 'inApp', not 'inapp'.
        defaultPreferences: {
          channels: {
            email: { enabled: true },
            sms:   { enabled: false },
            push:  { enabled: true },
            inApp: { enabled: true },
          },
        },
      },
    }),

    UsersModule,
  ],
})
export class AppModule {}

Producer side — emitting a notification

import { Service, Inject } from '@omnitron-dev/titan';
import { Public } from '@omnitron-dev/titan/netron';
import {
  NOTIFICATIONS_SERVICE,
  type NotificationsService,
} from '@omnitron-dev/titan-notifications';

@Service({ name: 'users' })
class UsersService {
  constructor(
    @Inject(NOTIFICATIONS_SERVICE) private readonly notify: NotificationsService,
  ) {}

  @Public()
  async create(input: CreateInput) {
    const user = await this.repo.create(input);

    // Returns immediately — actual delivery happens on a worker pod.
    // Recipient keys on `id` (not `userId`); payload needs `type`/`title`/`message`.
    await this.notify.send(
      { id: user.id, email: user.email },
      { type: 'announcement', title: 'Welcome', message: `Welcome, ${user.name}!`, data: { name: user.name } },
      { channels: ['email', 'inApp'] },
    );

    return user;
  }
}

Worker pod — consuming and delivering

import { Module, createToken } from '@omnitron-dev/titan';
import {
  NotificationsModule,
  type INotificationTargetResolver,
  type INotificationPersister,
  type INotificationRealtimeSignaler,
} from '@omnitron-dev/titan-notifications';

const USER_TARGET_RESOLVER_TOKEN = createToken<INotificationTargetResolver>('UserTargetResolver');
const NOTIFICATION_PERSISTER_TOKEN = createToken<INotificationPersister>('NotificationPersister');
const REALTIME_SIGNALER_TOKEN = createToken<INotificationRealtimeSignaler>('RealtimeSignaler');

@Module({
  imports: [
    ConfigModule.forRoot({/* … */}),
    LoggerModule.forRoot({ level: 'info' }),
    TitanRedisModule.forRoot({ config: { url: process.env.REDIS_URL } }),

    NotificationsModule.forWorker({
      targetResolver:   USER_TARGET_RESOLVER_TOKEN,
      persister:        NOTIFICATION_PERSISTER_TOKEN,
      realtimeSignaler: REALTIME_SIGNALER_TOKEN,
      workerOptions: {
        // There is no `concurrency`. The worker runs one XREADGROUP loop and
        // takes a BATCH per read; `readCount` is the knob that sizes it.
        readCount: 100,
        blockTimeoutMs: 5_000,
      },
    }),
  ],
  providers: [
    { provide: USER_TARGET_RESOLVER_TOKEN, useClass: MyUserResolver },
    { provide: NOTIFICATION_PERSISTER_TOKEN, useClass: MyPersister },
    { provide: REALTIME_SIGNALER_TOKEN, useClass: MyRealtimeSignaler },
  ],
})
export class WorkerModule {}

MyUserResolver, MyPersister, MyRealtimeSignaler are the three integration points you implement once per project. They wire the abstract worker to your concrete user store, your inbox storage, and your realtime channel.

A custom channel — e.g. SMTP email

import { AbstractEmailChannel, type EmailContent } from '@omnitron-dev/titan-notifications';
import nodemailer from 'nodemailer';

// AbstractEmailChannel fixes `name = 'email'` / `type = ChannelType.Email`
// and renders the payload; you only implement the transport.
class SmtpEmailChannel extends AbstractEmailChannel {
  private readonly transport = nodemailer.createTransport({/* … */});

  async sendEmail(to: string, content: EmailContent): Promise<{ messageId: string }> {
    const info = await this.transport.sendMail({
      to,
      subject: content.subject,
      html:    content.html,
      text:    content.text,
    });
    return { messageId: info.messageId };
  }
}

Register it via forRoot({ channels: [new SmtpEmailChannel()] }).

Cross-module wiring notes

Concern	Wiring detail
Producer vs worker mode	NotificationsModule.forRoot for producers, NotificationsModule.forWorker for workers — different DI shapes
Rotif transport	Configured under transport.rotif on the producer side; worker consumes via the same Redis instance
Rate-limiter location	RedisRateLimiter runs on the worker side, before each channel delivery
Preference enforcement	RedisPreferenceStore checked on the worker — producer doesn't know per-user prefs
Template caching	templates.cacheEnabled: true + cacheTTL avoid re-fetching template definitions on every send
Realtime + in-app	RealtimeSignaler pushes to WS/SSE; in-app channel persists the notification to the inbox
Webhook signature	webhookConfig.signatureSecret + signatureHeader for HMAC-signed delivery
DLQ	Failed deliveries land in a Rotif dead-letter queue (see Rotif config)
Redis retention	Off in every form unless configured — see the checklist item below

Production checklist

• Producer + worker pods deployed separately — independent scaling
• rateLimiterConfig.channelLimits.sms set — SMS is expensive and rate-limited by providers
• preferenceStoreConfig.defaultPreferences set — compliance with user opt-outs
• Templates registered at boot, not at first use
• DLQ monitored — repeated failures should alert
• Redis retention configured — every retention mechanism Rotif has is OFF by default, so a pipeline that never sets them grows in Redis memory for as long as it runs: - maxStreamLength (or minStreamId) — without one of these publish never trims the stream it writes to. The trim runs inside the publish script, so it is also the producer that bounds the stream: a stream stops being trimmed when publishing stops, at whatever size it reached. - dlqCleanup.enabled — defaults to false, and the sibling limits it gates (dlqCleanup.maxAge, dlqCleanup.maxSize) carry usable values that simply do not apply until it is turned on. A monitored DLQ is still an unbounded one. Sizing these is a deployment decision, but leaving all three unset is not a neutral default.
• Channel-specific timeouts — e.g. webhookConfig.timeout: 5_000
• Realtime signaler graceful degradation — if WS gateway is down, notification is still persisted
• Persistence backend backed up — in-app inbox should survive redis loss
• Test channels in staging — MockEmailChannel etc. for end-to-end verification

See also

• titan-notifications — full reference
• titan-events — in-process events the producer emits
• titan-ratelimit — alternative rate-limit backing
• titan-redis — Rotif transport runs on this