From 90e37edf890e50046a5c97ff3303125fd78be916 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Thu, 23 Jul 2026 15:47:40 +0530 Subject: [PATCH] feat(notifications): push for inbound mail, not just messenger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NotificationProjector now also consumes interaction.normalized and, for kind EMAIL (external email + app-to-app mail, which land via ingest — not message.sent), always notifies the non-sender recipient(s), reusing the same presence + mute gates. Refactors the fan-out into a shared notify() with an alwaysNotify flag (DM or mail); message.sent keeps its DM/mention/reply policy. Adds specs: mail notifies in a group thread where a plain message would not, and a non-EMAIL normalized interaction does not notify. Co-Authored-By: Claude Opus 4.8 --- .../notification.projector.spec.ts | 50 +++++++++++++--- .../notifications/notification.projector.ts | 57 +++++++++++++++---- 2 files changed, 89 insertions(+), 18 deletions(-) diff --git a/packages/iios-service/src/notifications/notification.projector.spec.ts b/packages/iios-service/src/notifications/notification.projector.spec.ts index bbe10a4..bd9acff 100644 --- a/packages/iios-service/src/notifications/notification.projector.spec.ts +++ b/packages/iios-service/src/notifications/notification.projector.spec.ts @@ -33,6 +33,18 @@ async function sentEvents(): Promise { const rows = await prisma.iiosOutboxEvent.findMany({ where: { eventType: IIOS_EVENTS.messageSent }, orderBy: { createdAt: 'asc' } }); return rows.map((r) => r.cloudEvent as unknown as CloudEvent); } +/** A synthetic interaction.normalized event (the shape ingest/mail emits) for an existing interaction. */ +function normalizedEvent(interactionId: string, threadId: string, kind: string): CloudEvent { + return { + specversion: '1.0', + id: `evt_norm_${interactionId}`, + type: IIOS_EVENTS.interactionNormalized, + source: 'iios/ingest/test', + time: '2026-01-01T00:00:00.000Z', + insignia: { idempotencyKey: `norm:${interactionId}` }, + data: { interactionId, threadId, kind }, + } as CloudEvent; +} function makeProjector(presence = new PresenceService(), deliverResult: 'sent' | 'gone' = 'sent') { const deliver = vi.fn().mockResolvedValue(deliverResult); const proj = new NotificationProjector(asService, new OutboxBus(), new DlqService(asService), new ProjectionCursorService(asService), presence, { deliver } as never); @@ -51,7 +63,7 @@ describe('NotificationProjector', () => { await seedSub('bob'); await m.send(threadId, alice, { content: 'hi bob' }, 'k1'); const { proj, deliver } = makeProjector(); - await proj.onMessageSent((await sentEvents())[0]!); + await proj.onEvent((await sentEvents())[0]!); expect(deliver).toHaveBeenCalledOnce(); expect(deliver.mock.calls[0][0].endpoint).toContain('push/bob'); }); @@ -63,7 +75,7 @@ describe('NotificationProjector', () => { await seedSub('bob'); await m.send(threadId, alice, { content: 'hello all' }, 'k1'); const { proj, deliver } = makeProjector(); - await proj.onMessageSent((await sentEvents())[0]!); + await proj.onEvent((await sentEvents())[0]!); expect(deliver).not.toHaveBeenCalled(); }); @@ -74,7 +86,7 @@ describe('NotificationProjector', () => { await seedSub('bob'); await m.send(threadId, alice, { content: 'hey @bob' }, 'k1', undefined, undefined, ['bob']); const { proj, deliver } = makeProjector(); - await proj.onMessageSent((await sentEvents())[0]!); + await proj.onEvent((await sentEvents())[0]!); expect(deliver).toHaveBeenCalledOnce(); }); @@ -88,7 +100,7 @@ describe('NotificationProjector', () => { await m.send(threadId, alice, { content: 'answer' }, 'k2', undefined, parent.id); // alice replies to bob (no mention) const { proj, deliver } = makeProjector(); const evs = await sentEvents(); - await proj.onMessageSent(evs[evs.length - 1]!); // project the reply + await proj.onEvent(evs[evs.length - 1]!); // project the reply expect(deliver).toHaveBeenCalledOnce(); expect(deliver.mock.calls[0][0].endpoint).toContain('push/bob'); }); @@ -102,7 +114,7 @@ describe('NotificationProjector', () => { const presence = new PresenceService(); presence.setFocus('sockB', 'bob', threadId); const { proj, deliver } = makeProjector(presence); - await proj.onMessageSent((await sentEvents())[0]!); + await proj.onEvent((await sentEvents())[0]!); expect(deliver).not.toHaveBeenCalled(); }); @@ -114,7 +126,31 @@ describe('NotificationProjector', () => { await prisma.iiosThreadParticipant.update({ where: { threadId_actorId: { threadId, actorId: bobActorId } }, data: { muted: true } }); await m.send(threadId, alice, { content: 'hi' }, 'k1'); const { proj, deliver } = makeProjector(); - await proj.onMessageSent((await sentEvents())[0]!); + await proj.onEvent((await sentEvents())[0]!); + expect(deliver).not.toHaveBeenCalled(); + }); + + it('mail (interaction.normalized, kind EMAIL): notifies the recipient even in a group thread', async () => { + // A group thread would NOT notify bob on message.sent (proven above); the EMAIL branch always does. + const m = ms(); + const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); + await m.addParticipant(threadId, alice, 'bob'); + await seedSub('bob'); + const sent = await m.send(threadId, alice, { content: 'your invoice is attached' }, 'k1'); + const { proj, deliver } = makeProjector(); + await proj.onEvent(normalizedEvent(sent.id, threadId, 'EMAIL')); + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver.mock.calls[0][0].endpoint).toContain('push/bob'); + }); + + it('interaction.normalized that is NOT mail (kind MESSAGE): does not notify', async () => { + const m = ms(); + const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); + await m.addParticipant(threadId, alice, 'bob'); + await seedSub('bob'); + const sent = await m.send(threadId, alice, { content: 'hello all' }, 'k1'); + const { proj, deliver } = makeProjector(); + await proj.onEvent(normalizedEvent(sent.id, threadId, 'MESSAGE')); expect(deliver).not.toHaveBeenCalled(); }); @@ -125,7 +161,7 @@ describe('NotificationProjector', () => { await seedSub('bob'); await m.send(threadId, alice, { content: 'hi' }, 'k1'); const { proj } = makeProjector(new PresenceService(), 'gone'); - await proj.onMessageSent((await sentEvents())[0]!); + await proj.onEvent((await sentEvents())[0]!); expect(await prisma.iiosNotificationSubscription.count()).toBe(0); }); }); diff --git a/packages/iios-service/src/notifications/notification.projector.ts b/packages/iios-service/src/notifications/notification.projector.ts index 366a725..5de27bd 100644 --- a/packages/iios-service/src/notifications/notification.projector.ts +++ b/packages/iios-service/src/notifications/notification.projector.ts @@ -14,14 +14,22 @@ interface MsgData { mentions?: string[]; } +interface NormalizedData { + interactionId: string; + threadId: string; + kind?: string; +} + /** - * Turns `message.sent` into push notifications for absent recipients. Three gates: - * 1. policy — DM always; group only if @mentioned or a reply to that recipient - * 2. presence — skip if the recipient is currently focused on that thread - * 3. mute — skip if the recipient muted the thread + * 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 is read - * from the opaque `membership` thread attribute here in the notification *policy*, not the kernel. + * 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 { @@ -37,31 +45,58 @@ export class NotificationProjector implements OnModuleInit { ) {} onModuleInit(): void { - this.dlq.registerHandler(this.consumer, (e) => this.onMessageSent(e)); + this.dlq.registerHandler(this.consumer, (e) => this.onEvent(e)); this.bus.on(IIOS_EVENTS.messageSent, (p) => - void this.onMessageSent(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err)), + 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 onMessageSent(event: CloudEvent): Promise { + 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'; - const mentions = data.mentions ?? []; // Parent author (for reply-to-you) — one lookup. let parentAuthorActorId: string | null = null; @@ -85,7 +120,7 @@ export class NotificationProjector implements OnModuleInit { // gate 1 — policy const mentioned = userId != null && mentions.includes(userId); const repliedToMe = parentAuthorActorId != null && parentAuthorActorId === p.actorId; - if (!(membership === 'dm' || mentioned || repliedToMe)) continue; + if (!(alwaysNotify || mentioned || repliedToMe)) continue; // gate 2 — presence if (userId && this.presence.isViewing(userId, data.threadId)) continue;