From 1cbfddd4c21e0706f378f70c6342dc2e660629a3 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Thu, 23 Jul 2026 15:52:53 +0530 Subject: [PATCH] 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); + } +}