From 1818bd874b90412e2419f282297c6f510469d44e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 6 Aug 2026 09:42:47 +0200 Subject: [PATCH] fix(02-11): authorize guardian egress under lock - Hold the managed-profile lock through every provider send\n- Add the notification event retention index and race coverage --- .../migration.sql | 5 ++ prisma/schema.prisma | 1 + .../notification-delivery-identity.test.ts | 8 +- src/lib/jobs/reminder/cleanup-handlers.ts | 8 +- src/lib/notifications/delivery-identity.ts | 83 +++++++++++++------ src/lib/notifications/dispatcher.ts | 25 ++++-- .../guardian-notification-fanout.test.ts | 28 ++++--- .../notification-attribution.test.ts | 38 ++++++++- 8 files changed, 145 insertions(+), 51 deletions(-) create mode 100644 prisma/migrations/0300_notification_event_created_at_index/migration.sql diff --git a/prisma/migrations/0300_notification_event_created_at_index/migration.sql b/prisma/migrations/0300_notification_event_created_at_index/migration.sql new file mode 100644 index 000000000..547ccd82b --- /dev/null +++ b/prisma/migrations/0300_notification_event_created_at_index/migration.sql @@ -0,0 +1,5 @@ +-- v1.37.0 — notification event retention deletes globally by created_at. +-- The dedup lookup index starts with record_user_id, so it cannot serve this +-- trailing-edge cleanup predicate efficiently. +CREATE INDEX "notification_events_created_at_idx" + ON "notification_events" ("created_at"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 153febee0..a8e052a70 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -5821,6 +5821,7 @@ model NotificationEvent { recordUser User @relation(fields: [recordUserId], references: [id], onDelete: Cascade) @@index([recordUserId, eventType, dedupKey, createdAt(sort: Desc)]) + @@index([createdAt], map: "notification_events_created_at_idx") @@map("notification_events") } diff --git a/src/__tests__/notification-delivery-identity.test.ts b/src/__tests__/notification-delivery-identity.test.ts index f29c3f9b2..c06892cc3 100644 --- a/src/__tests__/notification-delivery-identity.test.ts +++ b/src/__tests__/notification-delivery-identity.test.ts @@ -132,7 +132,13 @@ describe("notification delivery identity", () => { message: "Record schedule", }), ); - expect(transactionMock).toHaveBeenCalledTimes(1); + // Preflight chooses recipient channels, then the provider call is + // re-authorized under the managed-profile lock for the actual egress. + expect(transactionMock).toHaveBeenCalledTimes(2); + expect(transactionMock).toHaveBeenLastCalledWith(expect.any(Function), { + maxWait: 5_000, + timeout: 60_000, + }); }); it("refuses a managed record without an explicit recipient", async () => { diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts index 5216507e2..e5ae5775e 100644 --- a/src/lib/jobs/reminder/cleanup-handlers.ts +++ b/src/lib/jobs/reminder/cleanup-handlers.ts @@ -127,10 +127,10 @@ export async function handleMoodReminderCleanup( * notification events are claims rather than delivery diagnostics, but after * the same horizon their dedup windows have elapsed and they have no reader. * - * The DELETE is unbounded by user — the index covers `created_at` - * directly, so a `WHERE created_at < cutoff` scan is bounded by the - * size of the trailing-edge of the table rather than the live working - * set. On a one-million-row table with the documented retention + * The DELETE is unbounded by user. Both ledgers therefore carry a direct + * `created_at` index, so a `WHERE created_at < cutoff` scan is bounded by the + * size of the trailing edge rather than the live working set. On a + * one-million-row table with the documented retention * window, the daily prune touches ~11k rows (1M / 90d × 1d) and * completes in milliseconds. */ diff --git a/src/lib/notifications/delivery-identity.ts b/src/lib/notifications/delivery-identity.ts index cd82a8aac..961a49a44 100644 --- a/src/lib/notifications/delivery-identity.ts +++ b/src/lib/notifications/delivery-identity.ts @@ -80,13 +80,13 @@ export async function resolveManagedGuardianRecipientIds( /** * Resolve notification identities without treating a managed record as its * own destination. Legacy payloads remain valid only for ordinary self - * delivery; a managed record must name an active Guardian explicitly. + * delivery; a managed record must name an active Guardian explicitly. This is + * a preflight for recipient channel selection; managed delivery is authorized + * again under the same lock at each provider egress below. * - * Cross-principal authorization linearizes at the locked grant read below, - * immediately before the dispatcher selects a channel. A revocation that - * commits before that read prevents delivery; one that commits afterwards - * affects later dispatches but cannot recall provider work already in flight. - * The provider call deliberately runs outside the short database transaction. + * Cross-principal authorization linearizes at the locked grant read below. + * The egress helper keeps the same lock through the actual provider call, so + * a revocation cannot commit between Guardian authorization and delivery. */ export async function resolveNotificationDeliveryIdentity( payload: NotificationPayload, @@ -118,27 +118,60 @@ export async function resolveNotificationDeliveryIdentity( if (!isManagedGuardianFanoutAllowed(payload)) return null; - return prisma.$transaction(async (tx) => - withManagedProfileLock(tx, recordUserId, async (lockedRecord) => { - if (lockedRecord?.managedProfileAt == null) return null; + return withManagedGuardianAuthorization(recordUserId, recipientUserId, () => + Promise.resolve({ recordUserId, recipientUserId, managed: true }), + ); +} - const recipient = await tx.user.findUnique({ - where: { id: recipientUserId }, - select: { managedProfileAt: true }, - }); - if (!recipient || recipient.managedProfileAt != null) return null; +/** + * Execute a single managed Guardian channel send while holding the lifecycle + * advisory lock. The provider call intentionally remains inside this short + * transaction: managed Guardian revocation and expiry transitions take the + * same lock, so neither can commit after this authorization but before egress. + */ +export async function withManagedGuardianEgressAuthorization( + delivery: NotificationDeliveryIdentity, + send: () => Promise, +): Promise { + if (!delivery.managed) return send(); - const guardian = await tx.accountGrant.findFirst({ - where: { - grantorId: recordUserId, - granteeId: recipientUserId, - ...activeGuardianWhere(new Date()), - }, - select: { id: true }, - }); - if (!guardian) return null; + return withManagedGuardianAuthorization( + delivery.recordUserId, + delivery.recipientUserId, + send, + { maxWait: 5_000, timeout: 60_000 }, + ); +} - return { recordUserId, recipientUserId, managed: true }; - }), +async function withManagedGuardianAuthorization( + recordUserId: string, + recipientUserId: string, + operation: () => Promise, + options?: { maxWait: number; timeout: number }, +): Promise { + return prisma.$transaction( + (tx) => + withManagedProfileLock(tx, recordUserId, async (lockedRecord) => { + if (lockedRecord?.managedProfileAt == null) return null; + + const recipient = await tx.user.findUnique({ + where: { id: recipientUserId }, + select: { managedProfileAt: true }, + }); + if (!recipient || recipient.managedProfileAt != null) return null; + + const guardian = await tx.accountGrant.findFirst({ + where: { + grantorId: recordUserId, + granteeId: recipientUserId, + ...activeGuardianWhere(new Date()), + }, + select: { id: true }, + }); + if (!guardian) return null; + + return operation(); + }), + options, ); } diff --git a/src/lib/notifications/dispatcher.ts b/src/lib/notifications/dispatcher.ts index 751d56af5..ea1ac9c3f 100644 --- a/src/lib/notifications/dispatcher.ts +++ b/src/lib/notifications/dispatcher.ts @@ -33,6 +33,7 @@ import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-reco import { resolveManagedGuardianRecipientIds, resolveNotificationDeliveryIdentity, + withManagedGuardianEgressAuthorization, } from "@/lib/notifications/delivery-identity"; /** @@ -298,15 +299,25 @@ export async function dispatchNotification( continue; } - channelsAttempted += 1; - try { - const outcome = await sendToChannel( - channel.type as ChannelType, - channel.config, - delivery.recipientUserId, - recipientPayload, + const outcome = await withManagedGuardianEgressAuthorization( + delivery, + () => + sendToChannel( + channel.type as ChannelType, + channel.config, + delivery.recipientUserId, + recipientPayload, + ), ); + if (outcome === null) { + getEvent()?.addWarning( + "Guardian notification authorization changed before channel egress", + ); + continue; + } + + channelsAttempted += 1; if (outcome.ok) { await recordChannelSuccess({ diff --git a/tests/integration/guardian-notification-fanout.test.ts b/tests/integration/guardian-notification-fanout.test.ts index 22eb3f266..c8b17f563 100644 --- a/tests/integration/guardian-notification-fanout.test.ts +++ b/tests/integration/guardian-notification-fanout.test.ts @@ -33,11 +33,11 @@ vi.mock("@/lib/notifications/senders/ntfy", async () => { eventType: string; title: string; message: string; - }, - ) => { - ntfyEgress.onStart?.(); - if (ntfyEgress.block) await ntfyEgress.block; - const recipientUserId = payload.recipientUserId ?? payload.userId; + }, + ) => { + ntfyEgress.onStart?.(); + if (ntfyEgress.block) await ntfyEgress.block; + const recipientUserId = payload.recipientUserId ?? payload.userId; sent.ntfy.push({ recipientUserId, title: payload.title, @@ -402,10 +402,11 @@ describe("Guardian fan-out (real Postgres)", () => { addGuardian(record.id, reserveGuardian.id), addChannel(recipient.id, "NTFY"), ]); - const recipientGrant = await getPrismaClient().accountGrant.findFirstOrThrow({ - where: { grantorId: record.id, granteeId: recipient.id }, - select: { id: true }, - }); + const recipientGrant = + await getPrismaClient().accountGrant.findFirstOrThrow({ + where: { grantorId: record.id, granteeId: recipient.id }, + select: { id: true }, + }); let releaseEgress!: () => void; let markEgressStarted!: () => void; ntfyEgress.block = new Promise((resolve) => { @@ -446,10 +447,11 @@ describe("Guardian fan-out (real Postgres)", () => { addGuardian(record.id, reserveGuardian.id), addChannel(recipient.id, "NTFY"), ]); - const recipientGrant = await getPrismaClient().accountGrant.findFirstOrThrow({ - where: { grantorId: record.id, granteeId: recipient.id }, - select: { id: true }, - }); + const recipientGrant = + await getPrismaClient().accountGrant.findFirstOrThrow({ + where: { grantorId: record.id, granteeId: recipient.id }, + select: { id: true }, + }); let releaseEgress!: () => void; let markEgressStarted!: () => void; ntfyEgress.block = new Promise((resolve) => { diff --git a/tests/integration/notification-attribution.test.ts b/tests/integration/notification-attribution.test.ts index 75dd61e8d..ba6d50e16 100644 --- a/tests/integration/notification-attribution.test.ts +++ b/tests/integration/notification-attribution.test.ts @@ -1,7 +1,10 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { getPrismaClient, truncateAllTables } from "./setup"; -import { resolveNotificationDeliveryIdentity } from "@/lib/notifications/delivery-identity"; +import { + resolveNotificationDeliveryIdentity, + withManagedGuardianEgressAuthorization, +} from "@/lib/notifications/delivery-identity"; import { claimNotificationEvent } from "@/lib/notifications/reminder-dedup"; import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record"; @@ -154,6 +157,39 @@ describe("notification attribution (real Postgres)", () => { ); }); + it("refuses an expired Guardian after identity preflight but before provider egress", async () => { + const recordUser = await createUser("expiry-record", true); + const recipientUser = await createUser("expiry-recipient"); + const grant = await createGuardianGrant({ + recordUserId: recordUser.id, + recipientUserId: recipientUser.id, + expiresAt: new Date(Date.now() + 60_000), + }); + const delivery = await resolveNotificationDeliveryIdentity({ + eventType: "MEDICATION_REMINDER", + userId: recordUser.id, + recordUserId: recordUser.id, + recipientUserId: recipientUser.id, + title: "record reminder", + message: "record message", + }); + expect(delivery).toMatchObject({ managed: true }); + + await getPrismaClient().accountGrant.update({ + where: { id: grant.id }, + data: { expiresAt: new Date(Date.now() - 1) }, + }); + let sent = false; + + await expect( + withManagedGuardianEgressAuthorization(delivery!, async () => { + sent = true; + return { ok: true }; + }), + ).resolves.toBeNull(); + expect(sent).toBe(false); + }); + it("allows exactly one concurrent claim for a notification event", async () => { const recordUser = await createUser("claim-record", true); const client = getPrismaClient();