import { Inject, Injectable, OnModuleInit } from '@nestjs/common'; import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; import { DlqService } from '../outbox/dlq.service'; import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { PRESENCE_PORT, type PresencePort } from './presence.port'; import { NOTIFICATION_PORT, type NotificationPort } from './notification.port'; interface MsgData { interactionId: string; threadId: string; senderActorId: string; mentions?: string[]; } interface NormalizedData { interactionId: string; threadId: string; kind?: string; } /** * Turns messenger sends AND inbound mail into push notifications for absent recipients: * - `message.sent` → messenger policy (DM always; group only if @mentioned or reply-to-you) * - `interaction.normalized` (kind EMAIL) → mail always notifies the recipient (a directed 1:1) * Every candidate then passes two more gates before delivery: * presence — skip if the recipient is currently focused on that thread * mute — skip if the recipient muted the thread * Then dispatches to each of the recipient's subscriptions; a 'gone' result prunes it. * Idempotent per event id (same pattern as InboxProjector). Generic: DM-vs-group / mail is read * from opaque thread attributes here in the notification *policy*, not the kernel. */ @Injectable() export class NotificationProjector implements OnModuleInit { private readonly consumer = 'notification-projector'; constructor( private readonly prisma: PrismaService, private readonly bus: OutboxBus, private readonly dlq: DlqService, private readonly cursor: ProjectionCursorService, @Inject(PRESENCE_PORT) private readonly presence: PresencePort, @Inject(NOTIFICATION_PORT) private readonly port: NotificationPort, ) {} onModuleInit(): void { this.dlq.registerHandler(this.consumer, (e) => this.onEvent(e)); this.bus.on(IIOS_EVENTS.messageSent, (p) => void this.onEvent(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err)), ); // Mail (external email + app-to-app) lands via ingest, which emits interaction.normalized — not // message.sent — so subscribe here too and filter to EMAIL inside apply. this.bus.on(IIOS_EVENTS.interactionNormalized, (p) => void this.onEvent(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err)), ); } async onEvent(event: CloudEvent): Promise { if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance await this.apply(event); await this.cursor.advance(this.consumer, event); } private async apply(event: CloudEvent): Promise { if (event.type === IIOS_EVENTS.messageSent) return this.applyMessage(event); if (event.type === IIOS_EVENTS.interactionNormalized) return this.applyMail(event); } /** Messenger send: DM always notifies; group only on @mention or reply-to-you. */ private async applyMessage(event: CloudEvent): Promise { const data = event.data as MsgData; const thread = await this.prisma.iiosThread.findUnique({ where: { id: data.threadId } }); if (!thread) return; const membership = (thread.metadata as { membership?: string } | null)?.membership; await this.notify(data.threadId, data.interactionId, data.senderActorId, data.mentions ?? [], membership === 'dm'); } /** Inbound mail (kind EMAIL): a directed message — always notify the non-sender recipient(s). */ private async applyMail(event: CloudEvent): Promise { const data = event.data as NormalizedData; if (data.kind !== 'EMAIL') return; // other normalized interactions aren't mail — ignore const sender = await this.prisma.iiosInteraction.findUnique({ where: { id: data.interactionId }, select: { actorId: true } }); if (!sender?.actorId) return; await this.notify(data.threadId, data.interactionId, sender.actorId, [], true); } /** * Shared fan-out: load the message, then for every participant but the sender apply the policy * (alwaysNotify OR @mentioned OR replied-to-you), presence, and mute gates before delivering. */ private async notify(threadId: string, interactionId: string, senderActorId: string, mentions: string[], alwaysNotify: boolean): Promise { const data = { threadId, interactionId, senderActorId }; const interaction = await this.prisma.iiosInteraction.findUnique({ where: { id: data.interactionId }, include: { parts: { where: { kind: 'TEXT' }, take: 1 }, actor: { include: { sourceHandle: true } } }, }); const senderName = interaction?.actor?.sourceHandle?.externalId ?? 'Someone'; const body = interaction?.parts[0]?.bodyText ?? 'sent an attachment'; // Parent author (for reply-to-you) — one lookup. let parentAuthorActorId: string | null = null; if (interaction?.parentInteractionId) { const parent = await this.prisma.iiosInteraction.findUnique({ where: { id: interaction.parentInteractionId }, select: { actorId: true }, }); parentAuthorActorId = parent?.actorId ?? null; } const participants = await this.prisma.iiosThreadParticipant.findMany({ where: { threadId: data.threadId }, include: { actor: { include: { sourceHandle: true } } }, }); for (const p of participants) { if (p.actorId === data.senderActorId) continue; // never the sender const userId = p.actor?.sourceHandle?.externalId; // gate 1 — policy const mentioned = userId != null && mentions.includes(userId); const repliedToMe = parentAuthorActorId != null && parentAuthorActorId === p.actorId; if (!(alwaysNotify || mentioned || repliedToMe)) continue; // gate 2 — presence if (userId && (await this.presence.isViewing(userId, data.threadId))) continue; // gate 3 — mute if (p.muted) continue; const subs = await this.prisma.iiosNotificationSubscription.findMany({ where: { actorId: p.actorId } }); for (const s of subs) { const result = await this.port.deliver( { endpoint: s.endpoint, p256dh: s.p256dh, auth: s.auth }, { title: senderName, body, data: { threadId: data.threadId, interactionId: data.interactionId } }, ); if (result === 'gone') await this.prisma.iiosNotificationSubscription.delete({ where: { id: s.id } }); } } } private async claim(eventId: string): Promise { const res = await this.prisma.iiosProcessedEvent.createMany({ data: [{ consumerName: this.consumer, eventId }], skipDuplicates: true, }); return res.count > 0; } }