feat(iios): notification engine — presence-gated Web Push (engine phase)

Generic notification engine: reacts to message.sent and runs three gates before delivering — policy (DM always; group only on @mention or reply-to-you), presence (skip if the recipient is focused on that thread), mute (per-thread). Delivery via a swappable NotificationPort (Web Push/VAPID adapter); a 'gone' (404/410) prunes the sub. - PresenceService + gateway `focus_thread` signal + disconnect cleanup (room membership != viewing, since the sidebar joins every thread room). - IiosNotificationSubscription table + `muted` on IiosThreadParticipant (migration). - Endpoints: GET vapid-public-key, POST/DELETE subscribe, POST threads/:id/mute|unmute; listThreads returns the caller's `muted`. - DM-vs-group read from the opaque `membership` attribute in the notification POLICY only — kernel stays generic (grep-verified). Tests: 13 new (3 gates + reply-to-you + prune + web-push adapter states); full suite 205 green. Verified live: module boots, subscribe stored, mute/unmute round-trips. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-09 19:16:53 +05:30
parent f2ef8922ce
commit 0a8544b6fc
19 changed files with 646 additions and 5 deletions
@@ -29,7 +29,8 @@
    "prisma": "^6.2.1",
    "reflect-metadata": "^0.2.2",
    "rxjs": "^7.8.2",
-    "socket.io": "^4.8.3"
+    "socket.io": "^4.8.3",
+    "web-push": "^3.6.7"
  },
  "devDependencies": {
    "@insignia/iios-testkit": "workspace:*",
@@ -38,6 +39,7 @@
    "@types/express": "^5.0.6",
    "@types/jsonwebtoken": "^9.0.10",
    "@types/node": "^26.0.1",
+    "@types/web-push": "^3.6.4",
    "socket.io-client": "^4.8.3",
    "typescript": "^5.7.3"
  }
@@ -0,0 +1,30 @@
+-- AlterTable
+ALTER TABLE "IiosThreadParticipant" ADD COLUMN     "muted" BOOLEAN NOT NULL DEFAULT false;
+
+-- CreateTable
+CREATE TABLE "IiosNotificationSubscription" (
+    "id" TEXT NOT NULL,
+    "scopeId" TEXT NOT NULL,
+    "actorId" TEXT NOT NULL,
+    "kind" TEXT NOT NULL DEFAULT 'webpush',
+    "endpoint" TEXT NOT NULL,
+    "p256dh" TEXT NOT NULL,
+    "auth" TEXT NOT NULL,
+    "userAgent" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "lastSeenAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "IiosNotificationSubscription_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "IiosNotificationSubscription_endpoint_key" ON "IiosNotificationSubscription"("endpoint");
+
+-- CreateIndex
+CREATE INDEX "IiosNotificationSubscription_actorId_idx" ON "IiosNotificationSubscription"("actorId");
+
+-- AddForeignKey
+ALTER TABLE "IiosNotificationSubscription" ADD CONSTRAINT "IiosNotificationSubscription_scopeId_fkey" FOREIGN KEY ("scopeId") REFERENCES "IiosScope"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "IiosNotificationSubscription" ADD CONSTRAINT "IiosNotificationSubscription_actorId_fkey" FOREIGN KEY ("actorId") REFERENCES "IiosActorRef"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
@@ -309,6 +309,7 @@ model IiosScope {
  supportQueues IiosSupportQueue[]
  tickets       IiosTicket[]
  callbacks     IiosCallbackRequest[]
+  notificationSubscriptions IiosNotificationSubscription[]

  @@index([orgId, appId, tenantId])
 }
@@ -358,6 +359,7 @@ model IiosActorRef {
  ticketsRequested   IiosTicket[]            @relation("TicketRequester")
  ticketsAssigned    IiosTicket[]            @relation("TicketAssignee")
  callbacksRequested IiosCallbackRequest[]
+  notificationSubscriptions IiosNotificationSubscription[]
 }

 /// A configured channel surface (PORTAL in P1).
@@ -412,12 +414,32 @@ model IiosThreadParticipant {
  actorId         String
  actor           IiosActorRef @relation(fields: [actorId], references: [id])
  participantRole String       @default("MEMBER")
+  muted           Boolean      @default(false)
  joinedAt        DateTime     @default(now())
  leftAt          DateTime?

  @@id([threadId, actorId])
 }

+/// A device/browser push subscription for an actor (Web Push endpoint + keys).
+/// The notification engine delivers to these when the actor is absent.
+model IiosNotificationSubscription {
+  id         String       @id @default(cuid())
+  scopeId    String
+  scope      IiosScope    @relation(fields: [scopeId], references: [id], onDelete: Cascade)
+  actorId    String
+  actor      IiosActorRef @relation(fields: [actorId], references: [id])
+  kind       String       @default("webpush")
+  endpoint   String       @unique
+  p256dh     String
+  auth       String
+  userAgent  String?
+  createdAt  DateTime     @default(now())
+  lastSeenAt DateTime     @default(now())
+
+  @@index([actorId])
+}
+
 /// The semantic envelope. Idempotent per (scope, idempotencyKey).
 model IiosInteraction {
  id                  String              @id @default(cuid())
@@ -10,6 +10,7 @@ import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
 import { InboxModule } from './inbox/inbox.module';
 import { MediaModule } from './media/media.module';
+import { NotificationModule } from './notifications/notification.module';
 import { SupportModule } from './support/support.module';
 import { AdaptersModule } from './adapters/adapters.module';
 import { RoutingModule } from './routing/routing.module';
@@ -35,6 +36,7 @@ import { DevController } from './dev/dev.controller';
    MessageModule,
    InboxModule,
    MediaModule,
+    NotificationModule,
    SupportModule,
    AdaptersModule,
    RoutingModule,
@@ -4,6 +4,7 @@ import {
  ConnectedSocket,
  MessageBody,
  OnGatewayConnection,
+  OnGatewayDisconnect,
  OnGatewayInit,
  SubscribeMessage,
  WebSocketGateway,
@@ -15,6 +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';

 interface SocketState {
  principal: MessagePrincipal;
@@ -27,7 +29,7 @@ interface SocketState {
 * propagates messages persisted elsewhere (REST/ingest) to connected clients.
 */
@WebSocketGateway({ namespace: '/message', cors: { origin: true } })
-export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
+export class MessageGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
  private readonly logger = new Logger(MessageGateway.name);
  private readonly emittedLocally = new Set<string>();

@@ -37,6 +39,7 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
    private readonly messages: MessageService,
    private readonly session: SessionVerifier,
    private readonly bus: OutboxBus,
+    private readonly presence: PresenceService,
  ) {}

  afterInit(): void {
@@ -65,6 +68,18 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
    }
  }

+  handleDisconnect(client: Socket): void {
+    this.presence.clearSocket(client.id);
+  }
+
+  /** The app reports which thread is in the foreground (or null when blurred) — presence. */
+  @SubscribeMessage('focus_thread')
+  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);
+  }
+
  @SubscribeMessage('open_thread')
  async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string; membership?: string; creatorRole?: string; subject?: string }) {
    const { principal } = client.data as SocketState;
@@ -2,10 +2,11 @@ import { Module } from '@nestjs/common';
 import { MessageService } from './message.service';
 import { MessageGateway } from './message.gateway';
 import { OutboxModule } from '../outbox/outbox.module';
+import { PresenceService } from '../notifications/presence.service';

@Module({
  imports: [OutboxModule],
-  providers: [MessageService, MessageGateway],
-  exports: [MessageService],
+  providers: [MessageService, MessageGateway, PresenceService],
+  exports: [MessageService, PresenceService],
 })
 export class MessageModule {}
@@ -53,6 +53,7 @@ export interface ThreadSummary {
  participants: string[];
  participantCount: number;
  unread: number;
+  muted?: boolean;
  lastMessage?: string;
  lastAt?: Date;
 }
@@ -154,14 +155,27 @@ export class MessageService {
    return { threadId, participantCount: participantCount + 1 };
  }

+  /** Toggle the caller's per-thread notification mute flag. */
+  async muteThread(threadId: string, principal: MessagePrincipal, muted: boolean): Promise<{ threadId: string; muted: boolean }> {
+    const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
+    if (!thread) throw new NotFoundException('thread not found');
+    const actor = await this.actors.resolveActor(thread.scopeId, principal);
+    await this.prisma.iiosThreadParticipant.update({
+      where: { threadId_actorId: { threadId, actorId: actor.id } },
+      data: { muted },
+    });
+    return { threadId, muted };
+  }
+
  /** Generic "my threads": every thread the caller participates in, with last message + unread. */
  async listThreads(principal: MessagePrincipal): Promise<ThreadSummary[]> {
    const scope = await this.actors.findScope(principal);
    if (!scope) return [];
    const actor = await this.actors.resolveActor(scope.id, principal);
-    const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true } });
+    const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true, muted: true } });
    const threadIds = memberships.map((m) => m.threadId);
    if (threadIds.length === 0) return [];
+    const mutedBy = new Map(memberships.map((m) => [m.threadId, m.muted]));

    const [threads, unreads, allParts] = await Promise.all([
      this.prisma.iiosThread.findMany({ where: { id: { in: threadIds } } }),
@@ -193,6 +207,7 @@ export class MessageService {
          participants: members,
          participantCount: members.length,
          unread: unreadBy.get(t.id) ?? 0,
+          muted: mutedBy.get(t.id) ?? false,
          lastMessage: last?.parts[0]?.bodyText ?? undefined,
          lastAt: last?.occurredAt,
        };
@@ -0,0 +1,46 @@
+import { BadRequestException, Body, Controller, Delete, Get, Headers, Post } from '@nestjs/common';
+import { PrismaService } from '../prisma/prisma.service';
+import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
+import { SessionVerifier } from '../platform/session.verifier';
+import { SubscribeDto, UnsubscribeDto } from './notification.dto';
+
+@Controller('v1/notifications')
+export class NotificationController {
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly actors: ActorResolver,
+    private readonly session: SessionVerifier,
+  ) {}
+
+  /** The public VAPID key the client needs to create a push subscription. */
+  @Get('vapid-public-key')
+  vapidKey(): { key: string } {
+    return { key: process.env.VAPID_PUBLIC_KEY ?? '' };
+  }
+
+  /** Store (or refresh) a push subscription for the caller. */
+  @Post('subscribe')
+  async subscribe(@Body() body: SubscribeDto, @Headers('authorization') auth?: string): Promise<{ ok: true }> {
+    const principal = this.principal(auth);
+    const scope = await this.actors.resolveScope(principal);
+    const actor = await this.actors.resolveActor(scope.id, principal);
+    await this.prisma.iiosNotificationSubscription.upsert({
+      where: { endpoint: body.endpoint },
+      create: { scopeId: scope.id, actorId: actor.id, kind: body.kind ?? 'webpush', endpoint: body.endpoint, p256dh: body.keys.p256dh, auth: body.keys.auth, userAgent: body.userAgent },
+      update: { actorId: actor.id, p256dh: body.keys.p256dh, auth: body.keys.auth, lastSeenAt: new Date() },
+    });
+    return { ok: true };
+  }
+
+  @Delete('subscribe')
+  async unsubscribe(@Body() body: UnsubscribeDto): Promise<{ ok: true }> {
+    await this.prisma.iiosNotificationSubscription.deleteMany({ where: { endpoint: body.endpoint } });
+    return { ok: true };
+  }
+
+  private principal(auth?: string): MessagePrincipal {
+    const token = (auth ?? '').replace(/^Bearer\s+/i, '');
+    if (!token) throw new BadRequestException('Authorization bearer token is required');
+    return this.session.verify(token);
+  }
+}
@@ -0,0 +1,12 @@
+import { IsIn, IsObject, IsOptional, IsString } from 'class-validator';
+
+export class SubscribeDto {
+  @IsOptional() @IsIn(['webpush']) kind?: string;
+  @IsString() endpoint!: string;
+  @IsObject() keys!: { p256dh: string; auth: string };
+  @IsOptional() @IsString() userAgent?: string;
+}
+
+export class UnsubscribeDto {
+  @IsString() endpoint!: string;
+}
@@ -0,0 +1,26 @@
+import { Module } from '@nestjs/common';
+import { OutboxModule } from '../outbox/outbox.module';
+import { MessageModule } from '../messaging/message.module';
+import { NotificationController } from './notification.controller';
+import { NotificationProjector } from './notification.projector';
+import { WebPushDelivery } from './web-push.delivery';
+import { NOTIFICATION_PORT } from './notification.port';
+
+@Module({
+  imports: [OutboxModule, MessageModule], // OutboxModule → bus/dlq; MessageModule → PresenceService. Prisma/Projection/Identity are @Global.
+  controllers: [NotificationController],
+  providers: [
+    NotificationProjector,
+    {
+      // Dev binds Web Push (VAPID); prod can swap this to an email/FCM adapter.
+      provide: NOTIFICATION_PORT,
+      useFactory: () =>
+        new WebPushDelivery(
+          process.env.VAPID_PUBLIC_KEY
+            ? { publicKey: process.env.VAPID_PUBLIC_KEY, privateKey: process.env.VAPID_PRIVATE_KEY ?? '', subject: process.env.VAPID_SUBJECT ?? 'mailto:dev@insignia' }
+            : undefined,
+        ),
+    },
+  ],
+})
+export class NotificationModule {}
@@ -0,0 +1,18 @@
+/** The delivery seam. Dev = Web Push (VAPID); prod can swap to email/FCM/APNs. */
+export interface PushSub {
+  endpoint: string;
+  p256dh: string;
+  auth: string;
+}
+
+export interface NotificationPayload {
+  title: string;
+  body: string;
+  data: { threadId: string; interactionId?: string };
+}
+
+export interface NotificationPort {
+  deliver(sub: PushSub, payload: NotificationPayload): Promise<'sent' | 'gone' | 'failed'>;
+}
+
+export const NOTIFICATION_PORT = Symbol('NOTIFICATION_PORT');
@@ -0,0 +1,131 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
+import { makeFakePorts } from '@insignia/iios-testkit';
+import { resetDb } from '../test-utils/reset-db';
+import { MessageService, type MessagePrincipal } from '../messaging/message.service';
+import { ActorResolver } from '../identity/actor.resolver';
+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 { NotificationProjector } from './notification.projector';
+import type { PrismaService } from '../prisma/prisma.service';
+
+const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
+const prisma = new PrismaClient({ datasources: { db: { url } } });
+const asService = prisma as unknown as PrismaService;
+const actors = new ActorResolver(asService);
+const ms = () => new MessageService(asService, makeFakePorts(), actors);
+const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' };
+
+async function actorIdFor(userId: string): Promise<string> {
+  const handle = await prisma.iiosSourceHandle.findFirstOrThrow({ where: { externalId: userId } });
+  return (await prisma.iiosActorRef.findFirstOrThrow({ where: { sourceHandleId: handle.id } })).id;
+}
+async function seedSub(userId: string): Promise<string> {
+  const actorId = await actorIdFor(userId);
+  const scope = await prisma.iiosScope.findFirstOrThrow();
+  await prisma.iiosNotificationSubscription.create({ data: { scopeId: scope.id, actorId, kind: 'webpush', endpoint: `https://push/${userId}`, p256dh: 'k', auth: 'a' } });
+  return actorId;
+}
+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);
+}
+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);
+  return { proj, deliver, presence };
+}
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+describe('NotificationProjector', () => {
+  it('DM: notifies the recipient (absent), never the sender', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hi bob' }, 'k1');
+    const { proj, deliver } = makeProjector();
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).toHaveBeenCalledOnce();
+    expect(deliver.mock.calls[0][0].endpoint).toContain('push/bob');
+  });
+
+  it('group without a mention: 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');
+    await m.send(threadId, alice, { content: 'hello all' }, 'k1');
+    const { proj, deliver } = makeProjector();
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).not.toHaveBeenCalled();
+  });
+
+  it('group WITH a mention: notifies the mentioned member', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await m.addParticipant(threadId, alice, 'bob');
+    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]!);
+    expect(deliver).toHaveBeenCalledOnce();
+  });
+
+  it('group reply-to-you: notifies the parent author even without a mention', async () => {
+    const m = ms();
+    const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' };
+    const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    const parent = await m.send(threadId, bob, { content: 'question?' }, 'k1'); // bob authors the parent
+    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
+    expect(deliver).toHaveBeenCalledOnce();
+    expect(deliver.mock.calls[0][0].endpoint).toContain('push/bob');
+  });
+
+  it('presence gate: a recipient viewing the thread is NOT notified', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hi' }, 'k1');
+    const presence = new PresenceService();
+    presence.setFocus('sockB', 'bob', threadId);
+    const { proj, deliver } = makeProjector(presence);
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).not.toHaveBeenCalled();
+  });
+
+  it('mute gate: a muted recipient is NOT notified', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    const bobActorId = await seedSub('bob');
+    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]!);
+    expect(deliver).not.toHaveBeenCalled();
+  });
+
+  it('prunes a subscription that returns "gone"', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hi' }, 'k1');
+    const { proj } = makeProjector(new PresenceService(), 'gone');
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(await prisma.iiosNotificationSubscription.count()).toBe(0);
+  });
+});
@@ -0,0 +1,114 @@
+import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
+import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+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 { NOTIFICATION_PORT, type NotificationPort } from './notification.port';
+
+interface MsgData {
+  interactionId: string;
+  threadId: string;
+  senderActorId: string;
+  mentions?: 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
+ * 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.
+ */
+@Injectable()
+export class NotificationProjector implements OnModuleInit {
+  private readonly consumer = 'notification-projector';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly bus: OutboxBus,
+    private readonly dlq: DlqService,
+    private readonly cursor: ProjectionCursorService,
+    private readonly presence: PresenceService,
+    @Inject(NOTIFICATION_PORT) private readonly port: NotificationPort,
+  ) {}
+
+  onModuleInit(): void {
+    this.dlq.registerHandler(this.consumer, (e) => this.onMessageSent(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)),
+    );
+  }
+
+  async onMessageSent(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> {
+    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;
+
+    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;
+    if (interaction?.parentInteractionId) {
+      const parent = await this.prisma.iiosInteraction.findUnique({
+        where: { id: interaction.parentInteractionId },
+        select: { actorId: true },
+      });
+      parentAuthorActorId = parent?.actorId ?? null;
+    }
+
+    const participants = await this.prisma.iiosThreadParticipant.findMany({
+      where: { threadId: data.threadId },
+      include: { actor: { include: { sourceHandle: true } } },
+    });
+
+    for (const p of participants) {
+      if (p.actorId === data.senderActorId) continue; // never the sender
+      const userId = p.actor?.sourceHandle?.externalId;
+
+      // gate 1 — policy
+      const mentioned = userId != null && mentions.includes(userId);
+      const repliedToMe = parentAuthorActorId != null && parentAuthorActorId === p.actorId;
+      if (!(membership === 'dm' || mentioned || repliedToMe)) continue;
+
+      // gate 2 — presence
+      if (userId && this.presence.isViewing(userId, data.threadId)) continue;
+
+      // gate 3 — mute
+      if (p.muted) continue;
+
+      const subs = await this.prisma.iiosNotificationSubscription.findMany({ where: { actorId: p.actorId } });
+      for (const s of subs) {
+        const result = await this.port.deliver(
+          { endpoint: s.endpoint, p256dh: s.p256dh, auth: s.auth },
+          { title: senderName, body, data: { threadId: data.threadId, interactionId: data.interactionId } },
+        );
+        if (result === 'gone') await this.prisma.iiosNotificationSubscription.delete({ where: { id: s.id } });
+      }
+    }
+  }
+
+  private async claim(eventId: string): Promise<boolean> {
+    const res = await this.prisma.iiosProcessedEvent.createMany({
+      data: [{ consumerName: this.consumer, eventId }],
+      skipDuplicates: true,
+    });
+    return res.count > 0;
+  }
+}
@@ -0,0 +1,24 @@
+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', () => {
+    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);
+  });
+
+  it('any of the actor’s sockets counts as viewing', () => {
+    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);
+  });
+});
@@ -0,0 +1,25 @@
+import { Injectable } from '@nestjs/common';
+
+/**
+ * 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.
+ */
+@Injectable()
+export class PresenceService {
+  private readonly focus = new Map<string, { userId: string; threadId: string | null }>(); // socketId → focus
+
+  setFocus(socketId: string, userId: string, threadId: string | null): void {
+    this.focus.set(socketId, { userId, threadId });
+  }
+
+  clearSocket(socketId: string): void {
+    this.focus.delete(socketId);
+  }
+
+  isViewing(userId: string, threadId: string): boolean {
+    for (const f of this.focus.values()) if (f.userId === userId && f.threadId === threadId) return true;
+    return false;
+  }
+}
@@ -0,0 +1,34 @@
+import { describe, it, expect, vi } from 'vitest';
+import { WebPushDelivery } from './web-push.delivery';
+
+const sub = { endpoint: 'https://push/x', p256dh: 'k', auth: 'a' };
+const payload = { title: 'Bob', body: 'hi', data: { threadId: 'T1' } };
+const vapid = { publicKey: 'p', privateKey: 'k', subject: 'mailto:x' };
+
+describe('WebPushDelivery', () => {
+  it('returns "sent" on success', async () => {
+    const send = vi.fn().mockResolvedValue({ statusCode: 201 });
+    const d = new WebPushDelivery(vapid, send as never);
+    expect(await d.deliver(sub, payload)).toBe('sent');
+    expect(send).toHaveBeenCalledOnce();
+  });
+
+  it('returns "gone" on 404/410 so the caller prunes', async () => {
+    const d410 = new WebPushDelivery(vapid, vi.fn().mockRejectedValue({ statusCode: 410 }) as never);
+    expect(await d410.deliver(sub, payload)).toBe('gone');
+    const d404 = new WebPushDelivery(vapid, vi.fn().mockRejectedValue({ statusCode: 404 }) as never);
+    expect(await d404.deliver(sub, payload)).toBe('gone');
+  });
+
+  it('returns "failed" on other errors', async () => {
+    const d = new WebPushDelivery(vapid, vi.fn().mockRejectedValue({ statusCode: 500 }) as never);
+    expect(await d.deliver(sub, payload)).toBe('failed');
+  });
+
+  it('is disabled (returns "failed", never calls send) without VAPID keys', async () => {
+    const send = vi.fn();
+    const d = new WebPushDelivery(undefined, send as never);
+    expect(await d.deliver(sub, payload)).toBe('failed');
+    expect(send).not.toHaveBeenCalled();
+  });
+});
@@ -0,0 +1,40 @@
+import { Injectable } from '@nestjs/common';
+import webpush from 'web-push';
+import type { NotificationPayload, NotificationPort, PushSub } from './notification.port';
+
+type SendFn = typeof webpush.sendNotification;
+export interface Vapid {
+  publicKey: string;
+  privateKey: string;
+  subject: string;
+}
+
+/** Web Push (VAPID) delivery. A dead subscription (404/410) → 'gone' so the caller prunes it. */
+@Injectable()
+export class WebPushDelivery implements NotificationPort {
+  private readonly vapid?: Vapid;
+
+  constructor(
+    vapid?: Vapid,
+    private readonly send: SendFn = webpush.sendNotification,
+  ) {
+    // Store VAPID; pass it per-send (not global setVapidDetails) so construction never
+    // validates keys — keeps the adapter unit-testable with a mocked transport.
+    this.vapid = vapid?.publicKey ? vapid : undefined;
+  }
+
+  async deliver(sub: PushSub, payload: NotificationPayload): Promise<'sent' | 'gone' | 'failed'> {
+    if (!this.vapid) return 'failed'; // push disabled (no VAPID keys)
+    try {
+      await this.send(
+        { endpoint: sub.endpoint, keys: { p256dh: sub.p256dh, auth: sub.auth } },
+        JSON.stringify(payload),
+        { vapidDetails: this.vapid },
+      );
+      return 'sent';
+    } catch (e) {
+      const code = (e as { statusCode?: number }).statusCode;
+      return code === 404 || code === 410 ? 'gone' : 'failed';
+    }
+  }
+}
@@ -54,6 +54,19 @@ export class ThreadsController {
    return this.threads.getMessages(id);
  }

+  /** Mute / unmute notifications from this thread for the caller. */
+  @Post(':id/mute')
+  @HttpCode(200)
+  async mute(@Param('id') id: string, @Headers('authorization') auth?: string) {
+    return this.messages.muteThread(id, this.principal(auth), true);
+  }
+
+  @Post(':id/unmute')
+  @HttpCode(200)
+  async unmute(@Param('id') id: string, @Headers('authorization') auth?: string) {
+    return this.messages.muteThread(id, this.principal(auth), false);
+  }
+
  /** REST/polling fallback for native send (same write as the socket path). */
  @Post(':id/messages')
  @HttpCode(201)