feat(presence): Redis-backed presence for multi-replica prod

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

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],
+  providers: [MessageService, MessageGateway, presenceProvider],
-  exports: [MessageService, PresenceService],
+  exports: [MessageService, PRESENCE_PORT],
 })
 export class MessageModule {}

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;

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);
+    expect(await p.isViewing('alice', 'T1')).toBe(false);
-    p.setFocus('sock1', 'alice', 'T1');
+    await p.setFocus('sock1', 'alice', 'T1');
-    expect(p.isViewing('alice', 'T1')).toBe(true);
+    expect(await p.isViewing('alice', 'T1')).toBe(true);
-    expect(p.isViewing('alice', 'T2')).toBe(false); // joined-elsewhere ≠ viewing
+    expect(await p.isViewing('alice', 'T2')).toBe(false); // joined-elsewhere ≠ viewing
-    p.setFocus('sock1', 'alice', 'T2'); // moved focus
+    await p.setFocus('sock1', 'alice', 'T2'); // moved focus
-    expect(p.isViewing('alice', 'T1')).toBe(false);
+    expect(await p.isViewing('alice', 'T1')).toBe(false);
-    expect(p.isViewing('alice', 'T2')).toBe(true);
+    expect(await p.isViewing('alice', 'T2')).toBe(true);
-    p.clearSocket('sock1');
+    await p.clearSocket('sock1');
-    expect(p.isViewing('alice', 'T2')).toBe(false);
+    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');
+    await p.setFocus('sockA', 'bob', 'T9');
-    p.setFocus('sockB', 'bob', null); // a second tab, no focus
+    await p.setFocus('sockB', 'bob', null); // a second tab, no focus
-    expect(p.isViewing('bob', 'T9')).toBe(true);
+    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);
+  }
+}