import { Injectable, Logger } from '@nestjs/common'; import { NotificationKind, NotificationReadModel, SYSTEM_ACTOR_ID, SYSTEM_ACTOR_NAME, TrackerEvent, } from '@task-stream/api/tracker/domain'; import { ChangeFeed } from '../ports/change-feed'; import { NotificationMailer } from '../ports/notification-mailer'; import { NotificationsRepository } from '../ports/notifications.repository'; import { UserDirectory } from '../ports/user-directory'; import { versionOf } from '../event-handlers/read-model-notifier'; export interface NotificationDraft { userId: string; kind: NotificationKind; summary: string; } export interface NotificationContext { projectId: string; projectKey: string; issueId: string | null; issueKey: string | null; } /** * Delivers notifications: stored for the recipient, pushed live, and copied * by e-mail to whoever asked for it. Nobody is told about what they did * themselves, and each person is told once per event — the first reason wins. */ @Injectable() export class Notifier { private readonly logger = new Logger(Notifier.name); constructor( private readonly notifications: NotificationsRepository, private readonly changeFeed: ChangeFeed, private readonly mailer: NotificationMailer, private readonly users: UserDirectory, ) {} async deliver( event: TrackerEvent, context: NotificationContext, drafts: NotificationDraft[], ): Promise { const seen = new Set([event.actorId]); const wanted = drafts.filter((draft) => { if (seen.has(draft.userId)) return false; seen.add(draft.userId); return true; }); if (!wanted.length) return; const refs = await this.users.refs([event.actorId]); const actor = event.actorId === SYSTEM_ACTOR_ID ? { id: SYSTEM_ACTOR_ID, name: SYSTEM_ACTOR_NAME } : refs.of(event.actorId); await this.send( wanted.map((draft) => ({ id: `${event.aggregateId}:${versionOf(event)}:${draft.userId}`, userId: draft.userId, kind: draft.kind, ...context, actor, summary: draft.summary, readAt: null, createdAt: event.occurredAt, })), ); } /** * Stores, pushes and e-mails ready-made notifications. Their id makes them * idempotent: one that is already there is not sent again. */ async send(notifications: NotificationReadModel[]): Promise { const delivered: NotificationReadModel[] = []; for (const notification of notifications) { // Delivered again (at-least-once): already told, nothing to repeat. if (!(await this.notifications.insert(notification))) continue; await this.changeFeed.publishNotification(notification); delivered.push(notification); } if (this.mailer.enabled && delivered.length) { await this.email(delivered); } } /** A failed e-mail never blocks the event stream: it is logged and skipped. */ private async email(notifications: NotificationReadModel[]): Promise { const contacts = new Map( (await this.users.contacts(notifications.map((n) => n.userId))).map( (c) => [c.id, c], ), ); for (const notification of notifications) { const contact = contacts.get(notification.userId); if (!contact?.email) continue; if (!(await this.notifications.settings(notification.userId)).email) continue; try { await this.mailer.send( { name: contact.name, email: contact.email }, notification, ); } catch (error) { this.logger.warn( `E-mail to ${notification.userId} failed: ${(error as Error).message}`, ); } } } }