titan-notifications

official · @omnitron-dev/titan-notifications · stable

Multi-channel notification delivery (in-app, webhook, plus extensible email / SMS / push abstractions) with a template engine, Redis-backed rate limiting, per-user preferences, channel routing, scheduled and broadcast send, and a separate worker mode for consuming notification events from Rotif.

pnpm add @omnitron-dev/titan-notifications @omnitron-dev/titan-events @omnitron-dev/titan-redis

For SMTP email, use titan-mailer: a concrete Nodemailer channel, strict TLS, typed templates, bounded pools and a durable outbox worker over this module's Redis transport. Business-state changes and queued email are committed together; only opaque delivery IDs travel through Redis. SMTP acceptance is not an exactly-once or inbox-delivery guarantee.

When you need it

• Multi-channel delivery. Same payload, different channels per user (email + push for some, in-app only for others).

• Templated messages. Render once, send via any channel.

• Preference-respecting workflow. Users opt out of certain channels; rate limits stop you spamming.

• Producer/worker split. API pod publishes, worker pod consumes and delivers — forRoot and forWorker cover both ends.

Quickstart — producer (the app that emits notifications)

import { NotificationsModule } from '@omnitron-dev/titan-notifications';

@Module({
  imports: [
    TitanRedisModule.forRoot({ config: { url: env.REDIS_URL } }),
    NotificationsModule.forRoot({
      redis:           env.REDIS_URL,
      enableInApp:     true,
      enableWebhook:   true,
      defaultChannels: ['inapp', 'email'],
      templates:       { enabled: true, cacheEnabled: true },
      rateLimiterConfig: {
        defaultLimits: { perMinute: 10, perHour: 100, perDay: 1_000 },
        channelLimits: { sms: { perMinute: 1, perHour: 10 } },
      },
    }),
  ],
})
class AppModule {}

Quickstart — worker (the pod that consumes and delivers)

import { NotificationsModule } from '@omnitron-dev/titan-notifications';

@Module({
  imports: [
    NotificationsModule.forWorker({
      targetResolver:   NOTIFICATION_TARGET_RESOLVER,
      persister:        NOTIFICATION_PERSISTER,
      realtimeSignaler: NOTIFICATION_REALTIME_SIGNALER,
      // There is no `concurrency` — the worker runs one XREADGROUP loop and
      // takes a batch per read. `readCount` sizes the batch.
      workerOptions:    { readCount: 100, blockTimeoutMs: 5_000 },
    }),
  ],
  providers: [
    { provide: NOTIFICATION_TARGET_RESOLVER, useClass: MyUserResolver },
    { provide: NOTIFICATION_PERSISTER,        useClass: MyPersister },
    { provide: NOTIFICATION_REALTIME_SIGNALER,useClass: MyRealtimeSignaler },
  ],
})
class WorkerModule {}

The worker consumes events emitted by the producer (typically via Rotif), looks up channel targets, persists the notification, and signals realtime channels (WebSocket / SSE).

NotificationsModuleOptions

Also: forRootAsync({ useFactory, inject?, useExisting?, useClass?, imports?, isGlobal? }).

NotificationsService — the producer API

import { NotificationsService, NOTIFICATIONS_SERVICE }
  from '@omnitron-dev/titan-notifications';

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

  @Public()
  async welcome(user: User) {
    await this.notify.send(
      { id: user.id, email: user.email },
      { type: 'info', title: 'Welcome', message: `Hello, ${user.name}` },
      { channels: ['email', 'inapp'] },
    );
  }
}

NotificationPayload requires type, title and message. There is no template field on it — see below.

The service:

1. Checks rate limits per channel per recipient.

2. Checks preferences (does the user opt in to this channel?).

3. Routes to the right channel via IChannelRouter — only if one is registered; the injection is @Optional(), and without it the channels you pass in the options are used as given.

4. Hands off to the registered channel implementations.

5. Records the send against the rate limiter.

Templates are opt-in and caller-driven

send() does not render templates. TemplateEngine is a separate service exported under NOTIFICATIONS_TEMPLATE_ENGINE; nothing inside the package calls it. You apply it yourself and pass the result on:

import {
  NOTIFICATIONS_TEMPLATE_ENGINE, NOTIFICATIONS_SERVICE,
  TemplateEngine, NotificationsService,
} from '@omnitron-dev/titan-notifications';

@Service({ name: 'notify' })
class NotifyService {
  constructor(
    @Inject(NOTIFICATIONS_SERVICE) private readonly notify: NotificationsService,
    @Inject(NOTIFICATIONS_TEMPLATE_ENGINE) private readonly templates: TemplateEngine,
  ) {}

  @Public()
  async welcome(user: User) {
    const payload = this.templates.applyToPayload(
      'welcome',
      { type: 'info', title: '', message: '' },
      { name: user.name },
    );
    await this.notify.send({ id: user.id, email: user.email }, payload);
  }
}

applyToPayload overlays the rendered title/message onto the payload and stashes the full render under data._rendered. The templates block in the module options configures that engine (caching, TTL) — it does not switch on rendering inside send().

NotificationRecipient.id is the only required field, and it is load-bearing

interface NotificationRecipient {
  id: string;              // required
  email?: string; phone?: string; pushTokens?: string[];
  webhookUrl?: string; locale?: string;
}

Every step above keys off id: the rate-limit bucket (checkLimit(recipient.id, …)), the preference lookup, the realtime channel the delivery is published to (notifications.{type}.{recipient.id}), and the in-app channel's own canDeliver check, which is literally !!recipient.id.

That makes a misspelt key — userId, say — quiet rather than loud: the send resolves, in-app delivery is skipped for failing canDeliver, and everything else is published to notifications.welcome.undefined, where no subscriber is listening. Nothing throws. TypeScript is the only thing that catches it, so do not widen this argument to any.

Channels

Out-of-the-box implementations + extensible bases:

A concrete channel typically implements just deliver(recipient, rendered, options) plus optional validate().

Templates

NotificationsModule.forRoot({
  templates: { enabled: true, cacheEnabled: true, cacheTTL: 60_000 },
})

// Use:
await this.notify.send(
  { userId, email },
  { template: 'welcome', data: { name: 'Ada' } },
);

TemplateEngine supports variable substitution via {{name}} placeholders. Templates can be passed at boot or registered at runtime. DEFAULT_TEMPLATES is exported as a starting set.

Rate limiting and preferences

RedisRateLimiter enforces per-minute / per-hour / per-day limits globally or per-channel; RedisPreferenceStore tracks per-user opt-in/out per channel. Wired automatically when redis: is set.

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

preferenceStoreConfig: {
  defaultPreferences: { channels: { email: true, sms: false, push: true } },
}

Worker model

flowchart LR
  Api[API pod<br/>NotificationsService.send]
  Api -- emits notification event --> Rotif[(Rotif / queue)]
  Worker[Worker pod<br/>NotificationsModule.forWorker]
  Rotif --> Worker
  Worker --> Resolver[TargetResolver]
  Worker --> Persister[Persister<br/>store the notification]
  Worker --> Channel[Channel.deliver]
  Worker --> Realtime[RealtimeSignaler<br/>push to WS / SSE]

Producer/worker split lets you scale delivery independently from the request path and isolate flaky external providers (email, SMS) from the API tier.

@OnNotification decorator

import { OnNotification } from '@omnitron-dev/titan-notifications';

@Injectable()
class AuditListener {
  @OnNotification('notifications.notification.sent')
  async onSent(notification: any) { /* … */ }

  @OnNotification('notifications.notification.failed', { maxRetries: 5, exactlyOnce: true })
  async onFailed(notification: any) { /* … */ }
}

OnNotification(pattern, options?) takes the pattern as the first positional argument. options extends TransportSubscribeOptions (groupName, consumerName, maxRetries, retryDelay, retryStrategy, autoAck, prefetchCount, …) plus an exactlyOnce flag. Patterns are the transport topic names — the lifecycle topics are exported as NOTIFICATIONS_EVENTS (e.g. NOTIFICATIONS_EVENTS.NOTIFICATION_SENT === 'notifications.notification.sent').

Subscribe to delivery lifecycle events for observability, audit, metrics.

Health indicator

NotificationsHealthIndicator is exported and registers automatically with titan-health if both modules are loaded.

Observer failures also appear in the health report. Async lifecycle listeners are awaited; failures are logged and retained until the same event succeeds. An observer failure does not resend a notification that already reached its transport. Without an Events module, lifecycle emission is disabled and reported as such in the health details.

Operational diagnostics

@omnitron-dev/titan-notifications/diagnostics provides DiagnosticNotificationSink and DiagnosticNotificationsModule. They use the configured transport's queue, deduplication, retry and DLQ mechanisms. A missing consumer rejects delivery; broker acceptance is not recipient acknowledgement. See the contract, module wiring and outbox bridge.

Tokens (selection)

Lifecycle

The service is registered as a Titan provider; standard lifecycle hooks apply. The worker mode (forWorker) starts a Rotif consumer during the application onStart phase and stops it on onStop.

Anti-patterns

• Sending from the request path. Heavy channels (SMS / email) can take seconds. Send via the worker pattern, not directly in the request handler.

• Skipping preferences. Sending to users who opted out is a compliance risk in many jurisdictions. Always wire preferenceStore.

• Per-user templates. Templates are content; per-user data goes in the data payload. Pre-render with the engine, not in code.

• No rate limits. Without limits, a runaway loop can spam every user in your database in minutes.

See also

• titan-events — in-process events; this module bridges them to multi-channel delivery

• titan-redis — backs rate limiter, preference store, and Rotif

• titan-ratelimit — pluggable rate limiter for the producer side