From 745a23823a41c8f28f0a028d12e863c9c48d6580 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Thu, 23 Jul 2026 15:26:07 +0530 Subject: [PATCH 1/4] feat(messaging-ui): presence + live-unread adapter seams Add optional MessagingAdapter.setFocus(threadId) and subscribeActivity(cb) so hosts can report the foregrounded thread (backend suppresses push for it) and drive live conversation-list refresh on any thread's activity. Messenger wires focus/blur presence + activity-driven refetch. Published as 0.1.7. Co-Authored-By: Claude Opus 4.8 --- packages/iios-messaging-ui/package.json | 2 +- packages/iios-messaging-ui/src/adapter.ts | 9 ++++++++ .../src/components/messenger.tsx | 23 +++++++++++++++++++ 3 files changed, 33 insertions(+), 1 deletion(-) diff --git a/packages/iios-messaging-ui/package.json b/packages/iios-messaging-ui/package.json index 3274af2..0b7dcf3 100644 --- a/packages/iios-messaging-ui/package.json +++ b/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", diff --git a/packages/iios-messaging-ui/src/adapter.ts b/packages/iios-messaging-ui/src/adapter.ts index 3c72ace..0f7ca27 100644 --- a/packages/iios-messaging-ui/src/adapter.ts +++ b/packages/iios-messaging-ui/src/adapter.ts @@ -39,6 +39,15 @@ export interface MessagingAdapter { markRead(threadId: string, messageId: string): Promise; + /** 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. * diff --git a/packages/iios-messaging-ui/src/components/messenger.tsx b/packages/iios-messaging-ui/src/components/messenger.tsx index dd2effc..e9e0501 100644 --- a/packages/iios-messaging-ui/src/components/messenger.tsx +++ b/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]); From b8bf1aa347233cc3134f506ed71eb3fe9b1e66e7 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Thu, 23 Jul 2026 15:37:57 +0530 Subject: [PATCH 2/4] chore(env): document VAPID_* for Web Push offline notifications Web Push (VAPID) env for the already-built notification stack: unset VAPID_PUBLIC_KEY disables sends (WebPushDelivery returns 'failed'). Generate with 'npx web-push generate-vapid-keys'. Co-Authored-By: Claude Opus 4.8 --- .env.example | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.env.example b/.env.example index b99a182..fa35fec 100644 --- a/.env.example +++ b/.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" From 90e37edf890e50046a5c97ff3303125fd78be916 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Thu, 23 Jul 2026 15:47:40 +0530 Subject: [PATCH 3/4] 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; From 1cbfddd4c21e0706f378f70c6342dc2e660629a3 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Thu, 23 Jul 2026 15:52:53 +0530 Subject: [PATCH 4/4] feat(presence): Redis-backed presence for multi-replica prod MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract a PresencePort seam (async setFocus/clearSocket/isViewing) with two impls: the existing in-memory Map (dev, single instance) and a new RedisPresenceService (prod). MessageModule provides PRESENCE_PORT via a REDIS_URL-gated factory — same pattern as the socket.io Redis adapter — so 'who is viewing what' is shared across replicas and the notification projector's presence gate works at N>1. Gateway + projector inject the port; isViewing is now awaited. Presence spec updated to async (passing). Co-Authored-By: Claude Opus 4.8 --- .../src/messaging/message.gateway.spec.ts | 4 +- .../src/messaging/message.gateway.ts | 13 ++-- .../src/messaging/message.module.ts | 15 ++++- .../notifications/notification.projector.ts | 6 +- .../src/notifications/presence.port.ts | 15 +++++ .../notifications/presence.service.spec.ts | 28 ++++---- .../src/notifications/presence.service.ts | 14 ++-- .../notifications/redis-presence.service.ts | 64 +++++++++++++++++++ 8 files changed, 128 insertions(+), 31 deletions(-) create mode 100644 packages/iios-service/src/notifications/presence.port.ts create mode 100644 packages/iios-service/src/notifications/redis-presence.service.ts diff --git a/packages/iios-service/src/messaging/message.gateway.spec.ts b/packages/iios-service/src/messaging/message.gateway.spec.ts index 404ea8c..cfd1c35 100644 --- a/packages/iios-service/src/messaging/message.gateway.spec.ts +++ b/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, ); } diff --git a/packages/iios-service/src/messaging/message.gateway.ts b/packages/iios-service/src/messaging/message.gateway.ts index 1cef910..40bbe89 100644 --- a/packages/iios-service/src/messaging/message.gateway.ts +++ b/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') diff --git a/packages/iios-service/src/messaging/message.module.ts b/packages/iios-service/src/messaging/message.module.ts index a9578f3..cb1dac1 100644 --- a/packages/iios-service/src/messaging/message.module.ts +++ b/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 {} diff --git a/packages/iios-service/src/notifications/notification.projector.ts b/packages/iios-service/src/notifications/notification.projector.ts index 5de27bd..b4a2058 100644 --- a/packages/iios-service/src/notifications/notification.projector.ts +++ b/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 { @@ -40,7 +40,7 @@ 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, ) {} @@ -123,7 +123,7 @@ export class NotificationProjector implements OnModuleInit { 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; diff --git a/packages/iios-service/src/notifications/presence.port.ts b/packages/iios-service/src/notifications/presence.port.ts new file mode 100644 index 0000000..54d8dda --- /dev/null +++ b/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; + /** Forget everything about a socket (on disconnect). */ + clearSocket(socketId: string): Promise; + /** True if ANY of the user's sockets currently has this thread in the foreground. */ + isViewing(userId: string, threadId: string): Promise; +} + +export const PRESENCE_PORT = Symbol('PRESENCE_PORT'); diff --git a/packages/iios-service/src/notifications/presence.service.spec.ts b/packages/iios-service/src/notifications/presence.service.spec.ts index 5f94a19..e2028a7 100644 --- a/packages/iios-service/src/notifications/presence.service.spec.ts +++ b/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); }); }); diff --git a/packages/iios-service/src/notifications/presence.service.ts b/packages/iios-service/src/notifications/presence.service.ts index 0960275..0334e64 100644 --- a/packages/iios-service/src/notifications/presence.service.ts +++ b/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(); // socketId → focus - setFocus(socketId: string, userId: string, threadId: string | null): void { + async setFocus(socketId: string, userId: string, threadId: string | null): Promise { this.focus.set(socketId, { userId, threadId }); } - clearSocket(socketId: string): void { + async clearSocket(socketId: string): Promise { this.focus.delete(socketId); } - isViewing(userId: string, threadId: string): boolean { + async isViewing(userId: string, threadId: string): Promise { for (const f of this.focus.values()) if (f.userId === userId && f.threadId === threadId) return true; return false; } diff --git a/packages/iios-service/src/notifications/redis-presence.service.ts b/packages/iios-service/src/notifications/redis-presence.service.ts new file mode 100644 index 0000000..7050f3c --- /dev/null +++ b/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 { + 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 { + 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 { + const threads = await this.redis.hvals(this.userKey(userId)); + return threads.includes(threadId); + } + + async onModuleDestroy(): Promise { + await this.redis.quit().catch(() => undefined); + } +}