feat(notifications): push for inbound mail, not just messenger

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 <noreply@anthropic.com>

packages/iios-service/src/notifications/notification.projector.spec.ts

@@ -33,6 +33,18 @@ async function sentEvents(): Promise<CloudEvent[]> {
   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);
   });
 });

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:
+ * Turns messenger sends AND inbound mail into push notifications for absent recipients:
- *   1. policy   — DM always; group only if @mentioned or a reply to that recipient
+ *   - `message.sent`         → messenger policy (DM always; group only if @mentioned or reply-to-you)
- *   2. presence — skip if the recipient is currently focused on that thread
+ *   - `interaction.normalized` (kind EMAIL) → mail always notifies the recipient (a directed 1:1)
- *   3. mute     — skip if the recipient muted the thread
+ * 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
+ * Idempotent per event id (same pattern as InboxProjector). Generic: DM-vs-group / mail is read
- * from the opaque `membership` thread attribute here in the notification *policy*, not the kernel.
+ * 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<void> {
+  async onEvent(event: CloudEvent): Promise<void> {
     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<void> {
+    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<void> {
     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<void> {
+    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<void> {
+    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;