From 0a8544b6fcb3e4f9177cede128e4cd0210c0dc29 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Thu, 9 Jul 2026 19:16:53 +0530 Subject: [PATCH] =?UTF-8?q?feat(iios):=20notification=20engine=20=E2=80=94?= =?UTF-8?q?=20presence-gated=20Web=20Push=20(engine=20phase)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- packages/iios-service/package.json | 4 +- .../migration.sql | 30 ++++ packages/iios-service/prisma/schema.prisma | 22 +++ packages/iios-service/src/app.module.ts | 2 + .../src/messaging/message.gateway.ts | 17 ++- .../src/messaging/message.module.ts | 5 +- .../src/messaging/message.service.ts | 17 ++- .../notifications/notification.controller.ts | 46 ++++++ .../src/notifications/notification.dto.ts | 12 ++ .../src/notifications/notification.module.ts | 26 ++++ .../src/notifications/notification.port.ts | 18 +++ .../notification.projector.spec.ts | 131 ++++++++++++++++++ .../notifications/notification.projector.ts | 114 +++++++++++++++ .../notifications/presence.service.spec.ts | 24 ++++ .../src/notifications/presence.service.ts | 25 ++++ .../notifications/web-push.delivery.spec.ts | 34 +++++ .../src/notifications/web-push.delivery.ts | 40 ++++++ .../src/threads/threads.controller.ts | 13 ++ pnpm-lock.yaml | 71 ++++++++++ 19 files changed, 646 insertions(+), 5 deletions(-) create mode 100644 packages/iios-service/prisma/migrations/20260709133349_add_notifications/migration.sql create mode 100644 packages/iios-service/src/notifications/notification.controller.ts create mode 100644 packages/iios-service/src/notifications/notification.dto.ts create mode 100644 packages/iios-service/src/notifications/notification.module.ts create mode 100644 packages/iios-service/src/notifications/notification.port.ts create mode 100644 packages/iios-service/src/notifications/notification.projector.spec.ts create mode 100644 packages/iios-service/src/notifications/notification.projector.ts create mode 100644 packages/iios-service/src/notifications/presence.service.spec.ts create mode 100644 packages/iios-service/src/notifications/presence.service.ts create mode 100644 packages/iios-service/src/notifications/web-push.delivery.spec.ts create mode 100644 packages/iios-service/src/notifications/web-push.delivery.ts diff --git a/packages/iios-service/package.json b/packages/iios-service/package.json index 00b6c12..9705868 100644 --- a/packages/iios-service/package.json +++ b/packages/iios-service/package.json @@ -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" } diff --git a/packages/iios-service/prisma/migrations/20260709133349_add_notifications/migration.sql b/packages/iios-service/prisma/migrations/20260709133349_add_notifications/migration.sql new file mode 100644 index 0000000..e988d02 --- /dev/null +++ b/packages/iios-service/prisma/migrations/20260709133349_add_notifications/migration.sql @@ -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; diff --git a/packages/iios-service/prisma/schema.prisma b/packages/iios-service/prisma/schema.prisma index e535349..154896f 100644 --- a/packages/iios-service/prisma/schema.prisma +++ b/packages/iios-service/prisma/schema.prisma @@ -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()) diff --git a/packages/iios-service/src/app.module.ts b/packages/iios-service/src/app.module.ts index 3b48892..e12dc3d 100644 --- a/packages/iios-service/src/app.module.ts +++ b/packages/iios-service/src/app.module.ts @@ -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, diff --git a/packages/iios-service/src/messaging/message.gateway.ts b/packages/iios-service/src/messaging/message.gateway.ts index 7962dbb..3f62629 100644 --- a/packages/iios-service/src/messaging/message.gateway.ts +++ b/packages/iios-service/src/messaging/message.gateway.ts @@ -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(); @@ -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; diff --git a/packages/iios-service/src/messaging/message.module.ts b/packages/iios-service/src/messaging/message.module.ts index 765dff1..a9578f3 100644 --- a/packages/iios-service/src/messaging/message.module.ts +++ b/packages/iios-service/src/messaging/message.module.ts @@ -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 {} diff --git a/packages/iios-service/src/messaging/message.service.ts b/packages/iios-service/src/messaging/message.service.ts index 213aae2..f48a554 100644 --- a/packages/iios-service/src/messaging/message.service.ts +++ b/packages/iios-service/src/messaging/message.service.ts @@ -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 { 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, }; diff --git a/packages/iios-service/src/notifications/notification.controller.ts b/packages/iios-service/src/notifications/notification.controller.ts new file mode 100644 index 0000000..c69c6b5 --- /dev/null +++ b/packages/iios-service/src/notifications/notification.controller.ts @@ -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); + } +} diff --git a/packages/iios-service/src/notifications/notification.dto.ts b/packages/iios-service/src/notifications/notification.dto.ts new file mode 100644 index 0000000..d8c7034 --- /dev/null +++ b/packages/iios-service/src/notifications/notification.dto.ts @@ -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; +} diff --git a/packages/iios-service/src/notifications/notification.module.ts b/packages/iios-service/src/notifications/notification.module.ts new file mode 100644 index 0000000..c877516 --- /dev/null +++ b/packages/iios-service/src/notifications/notification.module.ts @@ -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 {} diff --git a/packages/iios-service/src/notifications/notification.port.ts b/packages/iios-service/src/notifications/notification.port.ts new file mode 100644 index 0000000..97cf1c9 --- /dev/null +++ b/packages/iios-service/src/notifications/notification.port.ts @@ -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'); diff --git a/packages/iios-service/src/notifications/notification.projector.spec.ts b/packages/iios-service/src/notifications/notification.projector.spec.ts new file mode 100644 index 0000000..bbe10a4 --- /dev/null +++ b/packages/iios-service/src/notifications/notification.projector.spec.ts @@ -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 { + 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 { + 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 { + 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); + }); +}); diff --git a/packages/iios-service/src/notifications/notification.projector.ts b/packages/iios-service/src/notifications/notification.projector.ts new file mode 100644 index 0000000..366a725 --- /dev/null +++ b/packages/iios-service/src/notifications/notification.projector.ts @@ -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 { + 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 { + 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 { + const res = await this.prisma.iiosProcessedEvent.createMany({ + data: [{ consumerName: this.consumer, eventId }], + skipDuplicates: true, + }); + return res.count > 0; + } +} diff --git a/packages/iios-service/src/notifications/presence.service.spec.ts b/packages/iios-service/src/notifications/presence.service.spec.ts new file mode 100644 index 0000000..5f94a19 --- /dev/null +++ b/packages/iios-service/src/notifications/presence.service.spec.ts @@ -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); + }); +}); diff --git a/packages/iios-service/src/notifications/presence.service.ts b/packages/iios-service/src/notifications/presence.service.ts new file mode 100644 index 0000000..0960275 --- /dev/null +++ b/packages/iios-service/src/notifications/presence.service.ts @@ -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(); // 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; + } +} diff --git a/packages/iios-service/src/notifications/web-push.delivery.spec.ts b/packages/iios-service/src/notifications/web-push.delivery.spec.ts new file mode 100644 index 0000000..c0c0b6b --- /dev/null +++ b/packages/iios-service/src/notifications/web-push.delivery.spec.ts @@ -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(); + }); +}); diff --git a/packages/iios-service/src/notifications/web-push.delivery.ts b/packages/iios-service/src/notifications/web-push.delivery.ts new file mode 100644 index 0000000..9460b90 --- /dev/null +++ b/packages/iios-service/src/notifications/web-push.delivery.ts @@ -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'; + } + } +} diff --git a/packages/iios-service/src/threads/threads.controller.ts b/packages/iios-service/src/threads/threads.controller.ts index 73ad083..58fb66f 100644 --- a/packages/iios-service/src/threads/threads.controller.ts +++ b/packages/iios-service/src/threads/threads.controller.ts @@ -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) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a8f0739..b58d8bd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -382,6 +382,9 @@ importers: socket.io: specifier: ^4.8.3 version: 4.8.3 + web-push: + specifier: ^3.6.7 + version: 3.6.7 devDependencies: '@insignia/iios-testkit': specifier: workspace:* @@ -401,6 +404,9 @@ importers: '@types/node': specifier: ^26.0.1 version: 26.0.1 + '@types/web-push': + specifier: ^3.6.4 + version: 3.6.4 socket.io-client: specifier: ^4.8.3 version: 4.8.3 @@ -1531,6 +1537,9 @@ packages: '@types/validator@13.15.10': resolution: {integrity: sha512-T8L6i7wCuyoK8A/ZeLYt1+q0ty3Zb9+qbSSvrIVitzT3YjZqkTZ40IbRsPanlB4h1QB3JVL1SYCdR6ngtFYcuA==} + '@types/web-push@3.6.4': + resolution: {integrity: sha512-GnJmSr40H3RAnj0s34FNTcJi1hmWFV5KXugE0mYWnYhgTAHLJ/dJKAwDmvPJYMke0RplY2XE9LnM4hqSqKIjhQ==} + '@types/ws@8.18.1': resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==} @@ -1639,6 +1648,10 @@ packages: engines: {node: '>=0.4.0'} hasBin: true + agent-base@7.1.4: + resolution: {integrity: sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ==} + engines: {node: '>= 14'} + ajv-formats@2.1.1: resolution: {integrity: sha512-Wx0Kx52hxE7C18hkMEggYlEifqWZtYaRgouJor+WMdPnQyEK13vgEWyVNup7SoeeoLMsr4kf5h6dOW11I15MUA==} peerDependencies: @@ -1702,6 +1715,9 @@ packages: array-timsort@1.0.3: resolution: {integrity: sha512-/+3GRL7dDAGEfM6TseQk/U+mi18TU2Ms9I3UlLdUMhz2hbvGNTKdj9xniwXfUqgYhHxRx0+8UnKkvlNwVU+cWQ==} + asn1.js@5.4.1: + resolution: {integrity: sha512-+I//4cYPccV8LdmBLiX8CYvf9Sp3vQsrqu2QNXRcrbiWvcx/UdlFiqUJJzxRQxgsZmvhXhn4cSKeSmoFjVdupA==} + assertion-error@2.0.1: resolution: {integrity: sha512-Izi8RQcffqCeNVgFigKli1ssklIbpHnCYc6AknXGYoB6grJqyeby7jv12JUQgmTAnIDnbck1uxksT4dzN3PWBA==} engines: {node: '>=12'} @@ -1728,6 +1744,9 @@ packages: bl@4.1.0: resolution: {integrity: sha512-1W07cM9gS6DcLperZfFSj+bWLtaPGSOHWhPiGzXmvVJbRLdG82sH/Kn8EtW1VqWVA54AKf2h5k5BbnIbwF3h6w==} + bn.js@4.12.5: + resolution: {integrity: sha512-3aRg6/JxfffFD+OlOjOFR3Vo79l39ooBTFucxx+MT3dhCtzn3EmiUPQo+6/OZuI2jbXi3YKgmiTFBgChQMwIRQ==} + body-parser@2.3.0: resolution: {integrity: sha512-2cGmJupaNgg+QUwVLAucDuWuoMZ6EX9iHDRswZ5lsNYEmwPaRknMPCLZz07yTzVq/83p4o/wzbDZbBrTvGGTIw==} engines: {node: '>=18'} @@ -2218,6 +2237,14 @@ packages: resolution: {integrity: sha512-4FbRdAX+bSdmo4AUFuS0WNiPz8NgFt+r8ThgNWmlrjQjt1Q7ZR9+zTlce2859x4KSXrwIsaeTqDoKQmtP8pLmQ==} engines: {node: '>= 0.8'} + http_ece@1.2.0: + resolution: {integrity: sha512-JrF8SSLVmcvc5NducxgyOrKXe3EsyHMgBFgSaIUGmArKe+rwr0uphRkRXvwiom3I+fpIfoItveHrfudL8/rxuA==} + engines: {node: '>=16'} + + https-proxy-agent@7.0.6: + resolution: {integrity: sha512-vK9P5/iUfdl95AI+JVyUuIcVtd4ofvtrOr3HNtM2yxC9bnMbEdp3x01OhQNnjb8IJYi38VlTE3mBXwcfvywuSw==} + engines: {node: '>= 14'} + iconv-lite@0.7.2: resolution: {integrity: sha512-im9DjEDQ55s9fL4EYzOAv0yMqmMBSZp6G0VvFyTMPKWxiSBHUj9NW/qqLmXUwXrrM7AvqSlTCfvqRb0cM8yYqw==} engines: {node: '>=0.10.0'} @@ -2428,6 +2455,9 @@ packages: resolution: {integrity: sha512-OqbOk5oEQeAZ8WXWydlu9HJjz9WVdEIvamMCcXmuqUYjTknH/sqsWvhQ3vgwKFRR1HpjvNBKQ37nbJgYzGqGcg==} engines: {node: '>=6'} + minimalistic-assert@1.0.1: + resolution: {integrity: sha512-UtJcAD4yEaGtjPezWuO9wC4nwUnVH/8/Im3yEHQP4b67cXlD/Qr9hdITCU1xDbSEXg2XKNaP8jsReV7vQd00/A==} + minimatch@10.2.5: resolution: {integrity: sha512-MULkVLfKGYDFYejP07QOurDLLQpcjk7Fw+7jXS2R2czRQzR56yHRveU5NDJEOviH+hETZKSkIk5c+T23GjFUMg==} engines: {node: 18 || 20 || >=22} @@ -3166,6 +3196,11 @@ packages: wcwidth@1.0.1: resolution: {integrity: sha512-XHPEwS0q6TaxcvG85+8EYkbiCux2XtWG2mkc47Ng2A77BQu9+DqIOJldST4HgPkuea7dvKSj5VgX3P1d4rW8Tg==} + web-push@3.6.7: + resolution: {integrity: sha512-OpiIUe8cuGjrj3mMBFWY+e4MMIkW3SVT+7vEIjvD9kejGUypv8GPDf84JdPWskK8zMRIJ6xYGm+Kxr8YkPyA0A==} + engines: {node: '>= 16'} + hasBin: true + webpack-node-externals@3.0.0: resolution: {integrity: sha512-LnL6Z3GGDPht/AigwRh2dvL9PQPFQ8skEpVrWZXLWBYmqcaojHNN0onvHzie6rq7EWKrrBfPYqNEzTJgiwEQDQ==} engines: {node: '>=6'} @@ -4148,6 +4183,10 @@ snapshots: '@types/validator@13.15.10': {} + '@types/web-push@3.6.4': + dependencies: + '@types/node': 26.0.1 + '@types/ws@8.18.1': dependencies: '@types/node': 26.0.1 @@ -4302,6 +4341,8 @@ snapshots: acorn@8.17.0: {} + agent-base@7.1.4: {} + ajv-formats@2.1.1(ajv@8.20.0): optionalDependencies: ajv: 8.20.0 @@ -4358,6 +4399,13 @@ snapshots: array-timsort@1.0.3: {} + asn1.js@5.4.1: + dependencies: + bn.js: 4.12.5 + inherits: 2.0.4 + minimalistic-assert: 1.0.1 + safer-buffer: 2.1.2 + assertion-error@2.0.1: {} balanced-match@1.0.2: {} @@ -4376,6 +4424,8 @@ snapshots: inherits: 2.0.4 readable-stream: 3.6.2 + bn.js@4.12.5: {} + body-parser@2.3.0: dependencies: bytes: 3.1.2 @@ -4960,6 +5010,15 @@ snapshots: statuses: 2.0.2 toidentifier: 1.0.1 + http_ece@1.2.0: {} + + https-proxy-agent@7.0.6: + dependencies: + agent-base: 7.1.4 + debug: 4.4.3 + transitivePeerDependencies: + - supports-color + iconv-lite@0.7.2: dependencies: safer-buffer: 2.1.2 @@ -5136,6 +5195,8 @@ snapshots: mimic-fn@2.1.0: {} + minimalistic-assert@1.0.1: {} + minimatch@10.2.5: dependencies: brace-expansion: 5.0.7 @@ -5847,6 +5908,16 @@ snapshots: dependencies: defaults: 1.0.4 + web-push@3.6.7: + dependencies: + asn1.js: 5.4.1 + http_ece: 1.2.0 + https-proxy-agent: 7.0.6 + jws: 4.0.1 + minimist: 1.2.8 + transitivePeerDependencies: + - supports-color + webpack-node-externals@3.0.0: {} webpack-sources@3.5.0: {}