.env.example

@@ -21,3 +21,10 @@ IIOS_CRED_KEY=""
 # Orphaned-media garbage collection (uploaded-but-never-sent attachments).
 # Interval 0 = off; set e.g. 3600000 (hourly) in prod. Grace defaults to 24h.
 IIOS_MEDIA_GC_INTERVAL_MS=0
+
+# Web Push (VAPID) for offline notifications. Unset VAPID_PUBLIC_KEY → push is disabled
+# (subscribe endpoint still stores subs, but WebPushDelivery returns 'failed' — no sends).
+# Generate a keypair with: npx web-push generate-vapid-keys
+VAPID_PUBLIC_KEY=""
+VAPID_PRIVATE_KEY=""
+VAPID_SUBJECT="mailto:dev@insignia"

packages/iios-messaging-ui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@insignia/iios-messaging-ui",
-  "version": "0.1.6",
+  "version": "0.1.7",
   "type": "module",
   "main": "dist/index.js",
   "module": "dist/index.js",

packages/iios-messaging-ui/src/adapter.ts

@@ -39,6 +39,15 @@ export interface MessagingAdapter {
 
   markRead(threadId: string, messageId: string): Promise<void>;
 
+  /** Report which thread is in the foreground (null when none/blurred) so the backend can suppress
+   *  push notifications for a thread you're actively viewing. Absent => presence isn't tracked. */
+  setFocus?(threadId: string | null): void;
+
+  /** Subscribe to activity across ALL of the caller's threads (not just the open one) — fires for
+   *  every incoming message. Drives live unread + in-app notifications. Absent => no cross-thread
+   *  awareness. Returns an unsubscribe fn. */
+  subscribeActivity?(cb: (e: { threadId: string; message: Message }) => void): Unsubscribe;
+
   /**
    * The current user's actor id, or null if not yet known.
    *

packages/iios-messaging-ui/src/components/messenger.tsx

@@ -40,6 +40,29 @@ export function Messenger({ focusThreadId }: { focusThreadId?: string | null } =
     setSelected(focusThreadId);
   }, [focusThreadId]);
 
+  // Presence: tell the backend which thread is foregrounded so it suppresses push for it. Null while
+  // browsing/composing, when the window is blurred, and on unmount (leaving the messenger).
+  useEffect(() => {
+    const active = browsing || composing ? null : selected;
+    adapter.setFocus?.(active);
+    const onBlur = (): void => adapter.setFocus?.(null);
+    const onFocus = (): void => adapter.setFocus?.(active);
+    window.addEventListener('blur', onBlur);
+    window.addEventListener('focus', onFocus);
+    return () => {
+      adapter.setFocus?.(null);
+      window.removeEventListener('blur', onBlur);
+      window.removeEventListener('focus', onFocus);
+    };
+  }, [adapter, selected, browsing, composing]);
+
+  // Live unread + ordering: refresh the conversation list when any of the caller's threads gets a
+  // message (not just the open one).
+  useEffect(() => {
+    if (!adapter.subscribeActivity) return;
+    return adapter.subscribeActivity(() => refetch());
+  }, [adapter, refetch]);
+
   const channels = useMemo(() => conversations.filter((c) => c.membership === 'channel'), [conversations]);
   const dms = useMemo(() => conversations.filter((c) => c.membership !== 'channel'), [conversations]);
   const selectedConversation = useMemo(() => conversations.find((c) => c.threadId === selected) ?? null, [conversations, selected]);

packages/iios-service/src/messaging/message.gateway.spec.ts

@@ -5,7 +5,7 @@ import { MessageGateway } from './message.gateway';
 import type { MessageService, MessagePrincipal } from './message.service';
 import { SessionVerifier } from '../platform/session.verifier';
 import type { OutboxBus } from '../outbox/outbox.bus';
-import type { PresenceService } from '../notifications/presence.service';
+import type { PresencePort } from '../notifications/presence.port';
 
 // Realtime delegation: the browser opens the socket with a short-lived token minted for the
 // realtime audience (iios-message). When IIOS_REALTIME_AUDIENCE is set, the gateway accepts ONLY
@@ -54,7 +54,7 @@ function gatewayReturning(principal: MessagePrincipal): MessageGateway {
     undefined as unknown as MessageService,
     session,
     undefined as unknown as OutboxBus,
-    undefined as unknown as PresenceService,
+    undefined as unknown as PresencePort,
   );
 }
 

packages/iios-service/src/messaging/message.gateway.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from 'node:crypto';
-import { Logger } from '@nestjs/common';
+import { Inject, Logger } from '@nestjs/common';
 import {
   ConnectedSocket,
   MessageBody,
@@ -16,7 +16,7 @@ import { logJson } from '../observability/logger';
 import { MessageService, type MessagePrincipal } from './message.service';
 import { SessionVerifier } from '../platform/session.verifier';
 import { OutboxBus } from '../outbox/outbox.bus';
-import { PresenceService } from '../notifications/presence.service';
+import { PRESENCE_PORT, type PresencePort } from '../notifications/presence.port';
 
 interface SocketState {
   principal: MessagePrincipal;
@@ -39,7 +39,7 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection, OnGat
     private readonly messages: MessageService,
     private readonly session: SessionVerifier,
     private readonly bus: OutboxBus,
-    private readonly presence: PresenceService,
+    @Inject(PRESENCE_PORT) private readonly presence: PresencePort,
   ) {}
 
   afterInit(): void {
@@ -77,7 +77,8 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection, OnGat
   }
 
   handleDisconnect(client: Socket): void {
-    this.presence.clearSocket(client.id);
+    // Fire-and-forget: presence is best-effort and must not block the disconnect path.
+    void this.presence.clearSocket(client.id).catch((err) => this.logger.warn(`presence clear failed: ${(err as Error).message}`));
   }
 
   /** The app reports which thread is in the foreground (or null when blurred) — presence. */
@@ -85,7 +86,9 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection, OnGat
   focusThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string | null }): void {
     const state = client.data as SocketState | undefined;
     if (!state?.principal) return;
-    this.presence.setFocus(client.id, state.principal.userId, body?.threadId ?? null);
+    void this.presence
+      .setFocus(client.id, state.principal.userId, body?.threadId ?? null)
+      .catch((err) => this.logger.warn(`presence set failed: ${(err as Error).message}`));
   }
 
   @SubscribeMessage('open_thread')

packages/iios-service/src/messaging/message.module.ts

@@ -3,10 +3,21 @@ import { MessageService } from './message.service';
 import { MessageGateway } from './message.gateway';
 import { OutboxModule } from '../outbox/outbox.module';
 import { PresenceService } from '../notifications/presence.service';
+import { RedisPresenceService } from '../notifications/redis-presence.service';
+import { PRESENCE_PORT } from '../notifications/presence.port';
+
+/**
+ * PRESENCE_PORT is single-instance in-memory by default; with REDIS_URL set it's Redis-backed so
+ * "who is viewing what" is shared across replicas (mirrors the socket.io Redis adapter gating).
+ */
+const presenceProvider = {
+  provide: PRESENCE_PORT,
+  useFactory: () => (process.env.REDIS_URL ? new RedisPresenceService(process.env.REDIS_URL) : new PresenceService()),
+};
 
 @Module({
   imports: [OutboxModule],
-  providers: [MessageService, MessageGateway, PresenceService],
-  exports: [MessageService, PresenceService],
+  providers: [MessageService, MessageGateway, presenceProvider],
+  exports: [MessageService, PRESENCE_PORT],
 })
 export class MessageModule {}

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

@@ -4,7 +4,7 @@ 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 { PresenceService } from './presence.service';
+import { PRESENCE_PORT, type PresencePort } from './presence.port';
 import { NOTIFICATION_PORT, type NotificationPort } from './notification.port';
 
 interface MsgData {
@@ -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 {
@@ -32,36 +40,63 @@ export class NotificationProjector implements OnModuleInit {
     private readonly bus: OutboxBus,
     private readonly dlq: DlqService,
     private readonly cursor: ProjectionCursorService,
-    private readonly presence: PresenceService,
+    @Inject(PRESENCE_PORT) private readonly presence: PresencePort,
     @Inject(NOTIFICATION_PORT) private readonly port: NotificationPort,
   ) {}
 
   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,10 +120,10 @@ 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;
+      if (userId && (await this.presence.isViewing(userId, data.threadId))) continue;
 
       // gate 3 — mute
       if (p.muted) continue;

packages/iios-service/src/notifications/presence.port.ts

@@ -0,0 +1,15 @@
+/**
+ * Presence seam: tracks which thread each socket has in the foreground so the notification
+ * projector can suppress push for a thread the recipient is actively viewing. Async so it can be
+ * backed by Redis across replicas (prod) or an in-memory Map on a single instance (dev).
+ */
+export interface PresencePort {
+  /** Record a socket's foregrounded thread (null when the window is blurred / no thread open). */
+  setFocus(socketId: string, userId: string, threadId: string | null): Promise<void>;
+  /** Forget everything about a socket (on disconnect). */
+  clearSocket(socketId: string): Promise<void>;
+  /** True if ANY of the user's sockets currently has this thread in the foreground. */
+  isViewing(userId: string, threadId: string): Promise<boolean>;
+}
+
+export const PRESENCE_PORT = Symbol('PRESENCE_PORT');

packages/iios-service/src/notifications/presence.service.spec.ts

@@ -2,23 +2,23 @@ import { describe, it, expect } from 'vitest';
 import { PresenceService } from './presence.service';
 
 describe('PresenceService', () => {
-  it('reports viewing only for the focused thread, and clears on disconnect', () => {
+  it('reports viewing only for the focused thread, and clears on disconnect', async () => {
     const p = new PresenceService();
-    expect(p.isViewing('alice', 'T1')).toBe(false);
-    p.setFocus('sock1', 'alice', 'T1');
-    expect(p.isViewing('alice', 'T1')).toBe(true);
-    expect(p.isViewing('alice', 'T2')).toBe(false); // joined-elsewhere ≠ viewing
-    p.setFocus('sock1', 'alice', 'T2'); // moved focus
-    expect(p.isViewing('alice', 'T1')).toBe(false);
-    expect(p.isViewing('alice', 'T2')).toBe(true);
-    p.clearSocket('sock1');
-    expect(p.isViewing('alice', 'T2')).toBe(false);
+    expect(await p.isViewing('alice', 'T1')).toBe(false);
+    await p.setFocus('sock1', 'alice', 'T1');
+    expect(await p.isViewing('alice', 'T1')).toBe(true);
+    expect(await p.isViewing('alice', 'T2')).toBe(false); // joined-elsewhere ≠ viewing
+    await p.setFocus('sock1', 'alice', 'T2'); // moved focus
+    expect(await p.isViewing('alice', 'T1')).toBe(false);
+    expect(await p.isViewing('alice', 'T2')).toBe(true);
+    await p.clearSocket('sock1');
+    expect(await p.isViewing('alice', 'T2')).toBe(false);
   });
 
-  it('any of the actor’s sockets counts as viewing', () => {
+  it('any of the actor’s sockets counts as viewing', async () => {
     const p = new PresenceService();
-    p.setFocus('sockA', 'bob', 'T9');
-    p.setFocus('sockB', 'bob', null); // a second tab, no focus
-    expect(p.isViewing('bob', 'T9')).toBe(true);
+    await p.setFocus('sockA', 'bob', 'T9');
+    await p.setFocus('sockB', 'bob', null); // a second tab, no focus
+    expect(await p.isViewing('bob', 'T9')).toBe(true);
   });
 });

packages/iios-service/src/notifications/presence.service.ts

@@ -1,24 +1,28 @@
 import { Injectable } from '@nestjs/common';
+import type { PresencePort } from './presence.port';
 
 /**
  * In-memory focus tracker (single instance). Keyed on userId (the stable externalId the
  * gateway has as principal.userId). NOTE: room membership ≠ viewing — the sidebar joins
  * every thread room for live updates, so presence uses an explicit `focus_thread` signal.
- * Prod (multi-replica): back this with Redis.
+ * Multi-replica prod uses {@link RedisPresenceService} instead (wired when REDIS_URL is set).
+ *
+ * Methods are async to satisfy the PresencePort seam; the map is mutated synchronously, so an
+ * un-awaited setFocus is still visible to an immediately-following isViewing.
  */
 @Injectable()
-export class PresenceService {
+export class PresenceService implements PresencePort {
   private readonly focus = new Map<string, { userId: string; threadId: string | null }>(); // socketId → focus
 
-  setFocus(socketId: string, userId: string, threadId: string | null): void {
+  async setFocus(socketId: string, userId: string, threadId: string | null): Promise<void> {
     this.focus.set(socketId, { userId, threadId });
   }
 
-  clearSocket(socketId: string): void {
+  async clearSocket(socketId: string): Promise<void> {
     this.focus.delete(socketId);
   }
 
-  isViewing(userId: string, threadId: string): boolean {
+  async isViewing(userId: string, threadId: string): Promise<boolean> {
     for (const f of this.focus.values()) if (f.userId === userId && f.threadId === threadId) return true;
     return false;
   }

packages/iios-service/src/notifications/redis-presence.service.ts

@@ -0,0 +1,64 @@
+import { Logger, type OnModuleDestroy } from '@nestjs/common';
+import { Redis } from 'ioredis';
+import type { PresencePort } from './presence.port';
+
+/**
+ * Redis-backed presence for multi-replica prod: a socket connected to replica A and a message
+ * projected on replica B must share the same "who is viewing what" view, which an in-memory Map
+ * cannot provide. Wired (in place of {@link PresenceService}) only when REDIS_URL is set.
+ *
+ * Layout (per user, tiny):
+ *   sock:{socketId}  -> userId            (so clearSocket can find the user), EX TTL
+ *   user:{userId}    -> HASH socketId=threadId, EX TTL
+ * isViewing = does the user hash hold this threadId for any socket. Keys carry a TTL so a crashed
+ * replica's entries self-heal; expiry errs toward "not viewing" (push is sent), the safe default.
+ */
+export class RedisPresenceService implements PresencePort, OnModuleDestroy {
+  private readonly logger = new Logger(RedisPresenceService.name);
+  private readonly redis: Redis;
+  private static readonly TTL_SECONDS = 300;
+  private static readonly PREFIX = 'iios:presence';
+
+  constructor(url: string, client?: Redis) {
+    this.redis = client ?? new Redis(url, { maxRetriesPerRequest: null });
+    this.redis.on('error', (err) => this.logger.error(`redis presence error: ${err.message}`));
+  }
+
+  private sockKey(socketId: string): string {
+    return `${RedisPresenceService.PREFIX}:sock:${socketId}`;
+  }
+  private userKey(userId: string): string {
+    return `${RedisPresenceService.PREFIX}:user:${userId}`;
+  }
+
+  async setFocus(socketId: string, userId: string, threadId: string | null): Promise<void> {
+    const ttl = RedisPresenceService.TTL_SECONDS;
+    const sock = this.sockKey(socketId);
+    const user = this.userKey(userId);
+    const pipe = this.redis.multi().set(sock, userId, 'EX', ttl);
+    if (threadId === null) {
+      // Blurred / no thread open — the socket stays connected but is viewing nothing.
+      pipe.hdel(user, socketId);
+    } else {
+      pipe.hset(user, socketId, threadId).expire(user, ttl);
+    }
+    await pipe.exec();
+  }
+
+  async clearSocket(socketId: string): Promise<void> {
+    const sock = this.sockKey(socketId);
+    const userId = await this.redis.get(sock);
+    const pipe = this.redis.multi().del(sock);
+    if (userId) pipe.hdel(this.userKey(userId), socketId);
+    await pipe.exec();
+  }
+
+  async isViewing(userId: string, threadId: string): Promise<boolean> {
+    const threads = await this.redis.hvals(this.userKey(userId));
+    return threads.includes(threadId);
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await this.redis.quit().catch(() => undefined);
+  }
+}