From 4b79f83cf7be45f4b773f9259bb8e189eb059ffe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 6 Aug 2026 13:46:42 +0200 Subject: [PATCH] fix(02-11): release guardian lock before provider egress - Commit a final authorization claim under the lifecycle lock\n- Keep provider calls and retry outcomes outside database transactions\n- Retain authorization claims with notification delivery diagnostics --- .../migration.sql | 32 ++++ prisma/schema.prisma | 28 +++ .../notification-delivery-identity.test.ts | 86 ++++++++- .../__tests__/notification-retention.test.ts | 9 +- src/lib/jobs/reminder/cleanup-handlers.ts | 19 +- src/lib/notifications/delivery-identity.ts | 137 ++++++++++---- src/lib/notifications/dispatcher.ts | 138 ++++++++++----- .../guardian-notification-fanout.test.ts | 167 +++++++++++++++++- .../notification-attribution.test.ts | 54 +++++- 9 files changed, 560 insertions(+), 110 deletions(-) create mode 100644 prisma/migrations/0301_notification_egress_authorization/migration.sql diff --git a/prisma/migrations/0301_notification_egress_authorization/migration.sql b/prisma/migrations/0301_notification_egress_authorization/migration.sql new file mode 100644 index 000000000..3ff6c1da4 --- /dev/null +++ b/prisma/migrations/0301_notification_egress_authorization/migration.sql @@ -0,0 +1,32 @@ +-- v1.37.0 — a managed Guardian notification receives a durable final +-- authorization claim under the profile lifecycle lock before provider I/O. +-- This records only delivery principals and outcome, never health content or +-- channel credentials. Retention follows the notification attempt cleanup. +CREATE TABLE "notification_egress_authorizations" ( + "id" TEXT NOT NULL, + "record_user_id" TEXT NOT NULL, + "recipient_user_id" TEXT NOT NULL, + "channel" TEXT NOT NULL, + "event_type" TEXT NOT NULL, + "authorized_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "completed_at" TIMESTAMP(3), + "outcome" TEXT, + + CONSTRAINT "notification_egress_authorizations_pkey" PRIMARY KEY ("id"), + CONSTRAINT "notification_egress_authorizations_record_user_id_fkey" + FOREIGN KEY ("record_user_id") REFERENCES "users"("id") + ON DELETE CASCADE ON UPDATE CASCADE, + CONSTRAINT "notification_egress_authorizations_recipient_user_id_fkey" + FOREIGN KEY ("recipient_user_id") REFERENCES "users"("id") + ON DELETE CASCADE ON UPDATE CASCADE +); + +CREATE INDEX "notification_egress_authorizations_authorized_at_idx" + ON "notification_egress_authorizations" ("authorized_at"); + +CREATE INDEX "notification_egress_authorizations_record_user_id_recipient_user_id_authorized_at_idx" + ON "notification_egress_authorizations" ( + "record_user_id", + "recipient_user_id", + "authorized_at" DESC + ); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index a8e052a70..021f06b0d 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -863,6 +863,8 @@ model User { pushAttemptsForRecord PushAttempt[] @relation("PushAttemptRecordUser") pushAttemptsReceived PushAttempt[] @relation("PushAttemptRecipientUser") notificationEvents NotificationEvent[] + notificationEgressAuthorizationsForRecord NotificationEgressAuthorization[] @relation("NotificationEgressAuthorizationRecordUser") + notificationEgressAuthorizationsReceived NotificationEgressAuthorization[] @relation("NotificationEgressAuthorizationRecipientUser") telegramReminderMessages TelegramReminderMessage[] // v1.10.3 — per-(user, day) Strain personal-anchor cache (day-total // TRIMP + running EWMA reference) so the nightly Strain engine anchors @@ -5825,6 +5827,32 @@ model NotificationEvent { @@map("notification_events") } +// A managed Guardian channel send crosses a revocable authorization boundary. +// The dispatcher creates this claim while it holds the managed-profile +// lifecycle lock, then commits before starting provider I/O. That committed +// claim is the linearization point: a revocation that commits first prevents +// the send, while a revocation that commits later does not cancel the one +// already-authorized provider operation. No notification content, channel +// credentials, or event deduplication keys are retained here. Completion is +// observational only; authorization and retry decisions never read it. +model NotificationEgressAuthorization { + id String @id @default(cuid()) + recordUserId String @map("record_user_id") + recipientUserId String @map("recipient_user_id") + channel String + eventType String @map("event_type") + authorizedAt DateTime @default(now()) @map("authorized_at") + completedAt DateTime? @map("completed_at") + outcome String? + + recordUser User @relation("NotificationEgressAuthorizationRecordUser", fields: [recordUserId], references: [id], onDelete: Cascade) + recipientUser User @relation("NotificationEgressAuthorizationRecipientUser", fields: [recipientUserId], references: [id], onDelete: Cascade) + + @@index([authorizedAt], map: "notification_egress_authorizations_authorized_at_idx") + @@index([recordUserId, recipientUserId, authorizedAt(sort: Desc)]) + @@map("notification_egress_authorizations") +} + // `result` is one of `"ok"` | `"error"` | `"skipped"` — `skipped` // covers the "no recipient" / "config missing" soft cases. // diff --git a/src/__tests__/notification-delivery-identity.test.ts b/src/__tests__/notification-delivery-identity.test.ts index c06892cc3..19698e453 100644 --- a/src/__tests__/notification-delivery-identity.test.ts +++ b/src/__tests__/notification-delivery-identity.test.ts @@ -1,6 +1,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const transactionMock = vi.fn(); +const authorizationClaimQueryMock = vi.fn(); +const authorizationUpdateMock = vi.fn(); vi.mock("@/lib/db", () => ({ prisma: { @@ -11,6 +13,9 @@ vi.mock("@/lib/db", () => ({ }, user: { findUnique: vi.fn() }, accountGrant: { findMany: vi.fn() }, + notificationEgressAuthorization: { + update: (...args: unknown[]) => authorizationUpdateMock(...args), + }, auditLog: { create: vi.fn() }, $transaction: (...args: unknown[]) => transactionMock(...args), }, @@ -66,6 +71,8 @@ const recipientUserId = "guardian-recipient"; beforeEach(() => { vi.resetAllMocks(); + authorizationClaimQueryMock.mockResolvedValue([{ id: "authorization" }]); + authorizationUpdateMock.mockResolvedValue({ id: "authorization" }); vi.mocked(prisma.user.findUnique).mockResolvedValue({ managedProfileAt: new Date(), } as never); @@ -85,7 +92,7 @@ beforeEach(() => { transactionMock.mockImplementation( async (operation: (tx: unknown) => Promise) => operation({ - $queryRaw: vi.fn(), + $queryRaw: (...args: unknown[]) => authorizationClaimQueryMock(...args), user: { findUnique: vi.fn(async ({ where }: { where: { id: string } }) => where.id === recordUserId @@ -132,12 +139,79 @@ describe("notification delivery identity", () => { message: "Record schedule", }), ); - // Preflight chooses recipient channels, then the provider call is - // re-authorized under the managed-profile lock for the actual egress. + // Preflight chooses recipient channels, then a short locked transaction + // records the final authorization claim before provider I/O starts. expect(transactionMock).toHaveBeenCalledTimes(2); - expect(transactionMock).toHaveBeenLastCalledWith(expect.any(Function), { - maxWait: 5_000, - timeout: 60_000, + expect(transactionMock).toHaveBeenLastCalledWith(expect.any(Function)); + expect(authorizationClaimQueryMock).toHaveBeenCalledTimes(3); + expect(authorizationUpdateMock).toHaveBeenCalledWith({ + where: { id: "authorization" }, + data: { completedAt: expect.any(Date), outcome: "ok" }, + }); + }); + + it("does not invoke a provider when final Guardian authorization is revoked", async () => { + let transactionCount = 0; + transactionMock.mockImplementation( + async (operation: (tx: unknown) => Promise) => { + transactionCount += 1; + return operation({ + $queryRaw: (...args: unknown[]) => + authorizationClaimQueryMock(...args), + user: { + findUnique: vi.fn(async ({ where }: { where: { id: string } }) => + where.id === recordUserId + ? { managedProfileAt: new Date() } + : { managedProfileAt: null }, + ), + }, + accountGrant: { + findFirst: vi.fn(async () => + transactionCount === 1 ? { id: "grant" } : null, + ), + }, + }); + }, + ); + + const outcome = await dispatchNotification({ + eventType: "MEDICATION_REMINDER", + userId: recordUserId, + recordUserId, + recipientUserId, + title: "Record content", + message: "Record schedule", + }); + + expect(outcome).toEqual({ + dispatched: false, + channelsAttempted: 0, + channelsSucceeded: 0, + }); + expect(authorizationClaimQueryMock).toHaveBeenCalledTimes(2); + expect(sendViaWebPushMock).not.toHaveBeenCalled(); + }); + + it("records a provider timeout on the claim without retaining the lifecycle lock", async () => { + sendViaWebPushMock.mockRejectedValue(new Error("provider timeout")); + + const outcome = await dispatchNotification({ + eventType: "MEDICATION_REMINDER", + userId: recordUserId, + recordUserId, + recipientUserId, + title: "Record content", + message: "Record schedule", + }); + + expect(outcome).toEqual({ + dispatched: false, + channelsAttempted: 1, + channelsSucceeded: 0, + }); + expect(authorizationUpdateMock).toHaveBeenCalledWith({ + where: { id: "authorization" }, + data: { completedAt: expect.any(Date), outcome: "sender_threw" }, }); }); diff --git a/src/lib/jobs/reminder/__tests__/notification-retention.test.ts b/src/lib/jobs/reminder/__tests__/notification-retention.test.ts index db4ba1443..2573d6698 100644 --- a/src/lib/jobs/reminder/__tests__/notification-retention.test.ts +++ b/src/lib/jobs/reminder/__tests__/notification-retention.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const h = vi.hoisted(() => ({ pushDeleteMany: vi.fn(), eventDeleteMany: vi.fn(), + authorizationDeleteMany: vi.fn(), meta: {} as Record, })); @@ -11,6 +12,7 @@ vi.mock("../shared", async (importOriginal) => ({ getWorkerPrisma: () => ({ pushAttempt: { deleteMany: h.pushDeleteMany }, notificationEvent: { deleteMany: h.eventDeleteMany }, + notificationEgressAuthorization: { deleteMany: h.authorizationDeleteMany }, }), })); @@ -37,10 +39,11 @@ beforeEach(() => { for (const key of Object.keys(h.meta)) delete h.meta[key]; h.pushDeleteMany.mockResolvedValue({ count: 3 }); h.eventDeleteMany.mockResolvedValue({ count: 5 }); + h.authorizationDeleteMany.mockResolvedValue({ count: 7 }); }); describe("notification retention", () => { - it("prunes record event anchors with the canonical bounded attempt retention", async () => { + it("prunes notification ledgers with the canonical bounded attempt retention", async () => { await handlePushAttemptCleanup([]); expect(h.pushDeleteMany).toHaveBeenCalledWith({ @@ -49,7 +52,11 @@ describe("notification retention", () => { expect(h.eventDeleteMany).toHaveBeenCalledWith({ where: { createdAt: { lt: expect.any(Date) } }, }); + expect(h.authorizationDeleteMany).toHaveBeenCalledWith({ + where: { authorizedAt: { lt: expect.any(Date) } }, + }); expect(h.meta.push_attempt_cleanup_deleted).toBe(3); expect(h.meta.notification_event_cleanup_deleted).toBe(5); + expect(h.meta.notification_egress_authorization_cleanup_deleted).toBe(7); }); }); diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts index e5ae5775e..e793bce69 100644 --- a/src/lib/jobs/reminder/cleanup-handlers.ts +++ b/src/lib/jobs/reminder/cleanup-handlers.ts @@ -127,9 +127,9 @@ 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. 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 + * The DELETE is unbounded by user. The event and attempt ledgers carry direct + * `created_at` indexes, and final egress claims carry `authorized_at`, so each + * trailing-edge scan is bounded by the size of 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. @@ -153,9 +153,20 @@ export async function handlePushAttemptCleanup( const deletedEvents = await p.notificationEvent.deleteMany({ where: { createdAt: { lt: cutoff } }, }); - const deleted = deletedAttempts.count + deletedEvents.count; + const deletedAuthorizations = + await p.notificationEgressAuthorization.deleteMany({ + where: { authorizedAt: { lt: cutoff } }, + }); + const deleted = + deletedAttempts.count + + deletedEvents.count + + deletedAuthorizations.count; evt.addMeta("push_attempt_cleanup_deleted", deletedAttempts.count); evt.addMeta("notification_event_cleanup_deleted", deletedEvents.count); + evt.addMeta( + "notification_egress_authorization_cleanup_deleted", + deletedAuthorizations.count, + ); return jobDone({ deleted }); } catch (err) { evt.addWarning(`push-attempt-cleanup failed: ${err}`); diff --git a/src/lib/notifications/delivery-identity.ts b/src/lib/notifications/delivery-identity.ts index 961a49a44..97c9710f0 100644 --- a/src/lib/notifications/delivery-identity.ts +++ b/src/lib/notifications/delivery-identity.ts @@ -1,10 +1,13 @@ +import { randomUUID } from "node:crypto"; + import { prisma } from "@/lib/db"; import { activeGuardianWhere, withManagedProfileLock, } from "@/lib/managed-profiles/lifecycle"; +import type { Prisma } from "@/generated/prisma/client"; import type { - ManagedGuardianFanoutEvent, + ChannelType, NotificationPayload, } from "@/lib/notifications/types"; @@ -82,11 +85,11 @@ export async function resolveManagedGuardianRecipientIds( * own destination. Legacy payloads remain valid only for ordinary self * 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. + * again under the same lock at each provider egress claim below. * - * 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. + * Cross-principal authorization preflight uses the locked grant read below. + * The final authorization is a separate short transaction that commits a + * durable egress claim immediately before the provider operation starts. */ export async function resolveNotificationDeliveryIdentity( payload: NotificationPayload, @@ -124,54 +127,112 @@ export async function resolveNotificationDeliveryIdentity( } /** - * 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. + * Create the final authorization claim for one managed Guardian channel. + * + * The claim commits while holding the same lifecycle lock as Guardian + * revocation and expiry. Its commit is the linearization point: if a + * revocation commits first, this returns null and no provider operation may + * start. If the claim commits first, it authorizes one provider operation + * that is already in flight from the lifecycle's perspective. The provider + * operation runs only after this transaction commits, so slow networks never + * retain a database transaction, advisory lock, or pooled connection. This is + * the logical send time, not a promise that a later revocation or expiry can + * recall an already-authorized provider operation before its first byte. */ -export async function withManagedGuardianEgressAuthorization( +export async function claimManagedGuardianEgressAuthorization( delivery: NotificationDeliveryIdentity, - send: () => Promise, -): Promise { - if (!delivery.managed) return send(); + channel: ChannelType, + eventType: string, +): Promise<{ id: string } | null> { + if (!delivery.managed) return null; return withManagedGuardianAuthorization( delivery.recordUserId, delivery.recipientUserId, - send, - { maxWait: 5_000, timeout: 60_000 }, + async (tx) => { + const id = randomUUID(); + const claims = await tx.$queryRaw>` + INSERT INTO "notification_egress_authorizations" ( + "id", + "record_user_id", + "recipient_user_id", + "channel", + "event_type", + "authorized_at" + ) + SELECT + ${id}, + ${delivery.recordUserId}, + ${delivery.recipientUserId}, + ${channel}, + ${eventType}, + clock_timestamp() + WHERE EXISTS ( + SELECT 1 + FROM "account_grants" + WHERE "grantor_id" = ${delivery.recordUserId} + AND "grantee_id" = ${delivery.recipientUserId} + AND "access" = 'MANAGE'::"account_grant_access" + AND "accepted_at" IS NOT NULL + AND "revoked_at" IS NULL + AND ( + "expires_at" IS NULL + OR "expires_at" > clock_timestamp() + ) + ) + RETURNING "id" + `; + return claims[0] ?? null; + }, ); } +/** + * Record a completed provider operation outside the lifecycle transaction. + * This is observability only: authorization always rechecks the active grant, + * and every retry creates a new claim regardless of a prior outcome. + */ +export async function completeManagedGuardianEgressAuthorization( + authorizationId: string, + outcome: "ok" | "hard_reject" | "transient_failure" | "sender_threw", +): Promise { + try { + await prisma.notificationEgressAuthorization.update({ + where: { id: authorizationId }, + data: { completedAt: new Date(), outcome }, + }); + } catch { + // The claim is intentionally durable even if the best-effort completion + // marker cannot be written after an otherwise independent provider call. + } +} + async function withManagedGuardianAuthorization( recordUserId: string, recipientUserId: string, - operation: () => Promise, - options?: { maxWait: number; timeout: number }, + operation: (tx: Prisma.TransactionClient) => Promise, ): Promise { - return prisma.$transaction( - (tx) => - withManagedProfileLock(tx, recordUserId, async (lockedRecord) => { - if (lockedRecord?.managedProfileAt == null) return null; + 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 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; + const guardian = await tx.accountGrant.findFirst({ + where: { + grantorId: recordUserId, + granteeId: recipientUserId, + ...activeGuardianWhere(new Date()), + }, + select: { id: true }, + }); + if (!guardian) return null; - return operation(); - }), - options, + return operation(tx); + }), ); } diff --git a/src/lib/notifications/dispatcher.ts b/src/lib/notifications/dispatcher.ts index ea1ac9c3f..4bc30dff2 100644 --- a/src/lib/notifications/dispatcher.ts +++ b/src/lib/notifications/dispatcher.ts @@ -31,9 +31,10 @@ import { } from "@/lib/notifications/client-managed-apns"; import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record"; import { + claimManagedGuardianEgressAuthorization, + completeManagedGuardianEgressAuthorization, resolveManagedGuardianRecipientIds, resolveNotificationDeliveryIdentity, - withManagedGuardianEgressAuthorization, } from "@/lib/notifications/delivery-identity"; /** @@ -300,25 +301,54 @@ export async function dispatchNotification( } try { - const outcome = await withManagedGuardianEgressAuthorization( - delivery, - () => - sendToChannel( - channel.type as ChannelType, - channel.config, - delivery.recipientUserId, - recipientPayload, - ), + const prepared = prepareChannelSend( + channel.type as ChannelType, + channel.config, + delivery.recipientUserId, + recipientPayload, ); - if (outcome === null) { - getEvent()?.addWarning( - "Guardian notification authorization changed before channel egress", - ); - continue; + let outcome: SendOutcome; + if ("outcome" in prepared) { + outcome = prepared.outcome; + channelsAttempted += 1; + } else { + const authorization = delivery.managed + ? await claimManagedGuardianEgressAuthorization( + delivery, + channel.type as ChannelType, + recipientPayload.eventType, + ) + : null; + if (delivery.managed && authorization === null) { + getEvent()?.addWarning( + "Guardian notification authorization changed before channel egress", + ); + continue; + } + + channelsAttempted += 1; + try { + outcome = await prepared.send(); + } catch (err) { + if (authorization) { + await completeManagedGuardianEgressAuthorization( + authorization.id, + "sender_threw", + ); + } + throw err; + } + if (authorization) { + await completeManagedGuardianEgressAuthorization( + authorization.id, + outcome.ok + ? "ok" + : outcome.hardReject + ? "hard_reject" + : "transient_failure", + ); + } } - - channelsAttempted += 1; - if (outcome.ok) { await recordChannelSuccess({ id: channel.id, @@ -387,12 +417,20 @@ export async function dispatchNotification( }; } -async function sendToChannel( +type PreparedChannelSend = + { send: () => Promise } | { outcome: SendOutcome }; + +/** + * Decrypt and validate a recipient-owned channel before the final managed + * Guardian authorization claim. The returned closure is the only code that + * can begin provider I/O, and is invoked immediately after that claim commits. + */ +function prepareChannelSend( type: ChannelType, encryptedConfig: string, recipientUserId: string, payload: NotificationPayload, -): Promise { +): PreparedChannelSend { let decrypted: string; try { decrypted = decrypt(encryptedConfig); @@ -406,9 +444,11 @@ async function sendToChannel( // a key-rotation gap every channel would otherwise auto-disable for the // wrong reason. return { - ok: false, - hardReject: true, - reason: `${type.toLowerCase()}_config_decrypt_failed`, + outcome: { + ok: false, + hardReject: true, + reason: `${type.toLowerCase()}_config_decrypt_failed`, + }, }; } @@ -421,12 +461,14 @@ async function sendToChannel( getEvent()?.addWarning("Failed to parse Telegram channel config"); // Malformed config decodes the same on every retry — permanent. return { - ok: false, - hardReject: true, - reason: "telegram_config_parse_failed", + outcome: { + ok: false, + hardReject: true, + reason: "telegram_config_parse_failed", + }, }; } - return sendViaTelegram(config, payload); + return { send: () => sendViaTelegram(config, payload) }; } case "NTFY": { let config: NtfyChannelConfig; @@ -436,12 +478,14 @@ async function sendToChannel( getEvent()?.addWarning("Failed to parse ntfy channel config"); // Malformed config decodes the same on every retry — permanent. return { - ok: false, - hardReject: true, - reason: "ntfy_config_parse_failed", + outcome: { + ok: false, + hardReject: true, + reason: "ntfy_config_parse_failed", + }, }; } - return sendViaNtfy(config, payload); + return { send: () => sendViaNtfy(config, payload) }; } case "WEBHOOK": { let config: WebhookChannelConfig; @@ -451,12 +495,14 @@ async function sendToChannel( getEvent()?.addWarning("Failed to parse webhook channel config"); // Malformed config decodes the same on every retry — permanent. return { - ok: false, - hardReject: true, - reason: "webhook_config_parse_failed", + outcome: { + ok: false, + hardReject: true, + reason: "webhook_config_parse_failed", + }, }; } - return sendViaWebhook(config, payload); + return { send: () => sendViaWebhook(config, payload) }; } case "EMAIL": { let config: EmailChannelConfig; @@ -466,15 +512,17 @@ async function sendToChannel( getEvent()?.addWarning("Failed to parse email channel config"); // Malformed config decodes the same on every retry — permanent. return { - ok: false, - hardReject: true, - reason: "email_config_parse_failed", + outcome: { + ok: false, + hardReject: true, + reason: "email_config_parse_failed", + }, }; } - return sendViaEmail(config, payload); + return { send: () => sendViaEmail(config, payload) }; } case "WEB_PUSH": { - return sendViaWebPush(recipientUserId, payload); + return { send: () => sendViaWebPush(recipientUserId, payload) }; } case "APNS": { // APNs config lives on the Device row, not in the channel config. @@ -482,14 +530,16 @@ async function sendToChannel( // we just keep the symmetric path so future per-user overrides // (e.g. a custom APNs topic) can hang off the channel without // changing the dispatcher shape. - return sendViaApns(recipientUserId, payload); + return { send: () => sendViaApns(recipientUserId, payload) }; } default: getEvent()?.addWarning(`Unknown notification channel type: ${type}`); return { - ok: false, - hardReject: false, - reason: "unknown_channel_type", + outcome: { + ok: false, + hardReject: false, + reason: "unknown_channel_type", + }, }; } } diff --git a/tests/integration/guardian-notification-fanout.test.ts b/tests/integration/guardian-notification-fanout.test.ts index c8b17f563..efc49c218 100644 --- a/tests/integration/guardian-notification-fanout.test.ts +++ b/tests/integration/guardian-notification-fanout.test.ts @@ -17,6 +17,8 @@ const sent = vi.hoisted(() => ({ const ntfyEgress = vi.hoisted(() => ({ block: null as Promise | null, onStart: null as (() => void) | null, + error: null as Error | null, + outcome: null as { ok: false; reason: string } | null, })); vi.mock("@/lib/notifications/senders/ntfy", async () => { @@ -37,6 +39,8 @@ vi.mock("@/lib/notifications/senders/ntfy", async () => { ) => { ntfyEgress.onStart?.(); if (ntfyEgress.block) await ntfyEgress.block; + if (ntfyEgress.error) throw ntfyEgress.error; + if (ntfyEgress.outcome) return ntfyEgress.outcome; const recipientUserId = payload.recipientUserId ?? payload.userId; sent.ntfy.push({ recipientUserId, @@ -199,6 +203,8 @@ async function addChannel( beforeEach(async () => { ntfyEgress.block = null; ntfyEgress.onStart = null; + ntfyEgress.error = null; + ntfyEgress.outcome = null; sent.deliveries.length = 0; sent.ntfy.length = 0; sent.apns.length = 0; @@ -375,6 +381,22 @@ describe("Guardian fan-out (real Postgres)", () => { expect(attempts).toHaveLength(expected.length); expect(attempts).toEqual(expect.arrayContaining(expected)); }); + const claims = + await getPrismaClient().notificationEgressAuthorization.findMany({ + where: { eventType: "MEDICATION_REMINDER" }, + select: { + channel: true, + recordUserId: true, + recipientUserId: true, + eventType: true, + }, + }); + const expectedClaims = expected.map((claim) => ({ + ...claim, + eventType: "MEDICATION_REMINDER", + })); + expect(claims).toHaveLength(expectedClaims.length); + expect(claims).toEqual(expect.arrayContaining(expectedClaims)); await dispatchNotification({ eventType: "MEDICATION_REMINDER", @@ -391,7 +413,7 @@ describe("Guardian fan-out (real Postgres)", () => { ).resolves.toBe(expected.length); }); - it("keeps a Guardian revocation from committing while its channel egress is in flight", async () => { + it("allows Guardian revocation to commit while a claimed channel egress is in flight", async () => { const [record, recipient, reserveGuardian] = await Promise.all([ createUser({ label: "record", locale: "en", managed: true }), createUser({ label: "recipient", locale: "en" }), @@ -424,19 +446,27 @@ describe("Guardian fan-out (real Postgres)", () => { message: "record message", }); await egressStarted; + await expect( + getPrismaClient().notificationEgressAuthorization.count({ + where: { + recordUserId: record.id, + recipientUserId: recipient.id, + channel: "NTFY", + }, + }), + ).resolves.toBe(1); const revocation = revokeGrant({ grantId: recipientGrant.id, grantorId: record.id, }); - await expectPromiseToRemainPending(revocation); + await expectPromiseToSettle(revocation); releaseEgress(); await delivery; - await revocation; }); - it("keeps a Guardian expiry transition from committing while its channel egress is in flight", async () => { + it("allows Guardian expiry to commit while a claimed channel egress is in flight", async () => { const [record, recipient, reserveGuardian] = await Promise.all([ createUser({ label: "record", locale: "en", managed: true }), createUser({ label: "recipient", locale: "en" }), @@ -469,6 +499,15 @@ describe("Guardian fan-out (real Postgres)", () => { message: "record message", }); await egressStarted; + await expect( + getPrismaClient().notificationEgressAuthorization.count({ + where: { + recordUserId: record.id, + recipientUserId: recipient.id, + channel: "NTFY", + }, + }), + ).resolves.toBe(1); const expiry = getPrismaClient().$transaction((tx) => withManagedProfileLock(tx, record.id, () => @@ -478,21 +517,131 @@ describe("Guardian fan-out (real Postgres)", () => { }), ), ); - await expectPromiseToRemainPending(expiry); + await expectPromiseToSettle(expiry); releaseEgress(); await delivery; - await expiry; + }); + + it("records a timed-out provider after the authorization claim has committed", async () => { + const [record, recipient, reserveGuardian] = await Promise.all([ + createUser({ label: "record", locale: "en", managed: true }), + createUser({ label: "recipient", locale: "en" }), + createUser({ label: "reserve-guardian", locale: "de" }), + ]); + await Promise.all([ + addGuardian(record.id, recipient.id), + addGuardian(record.id, reserveGuardian.id), + addChannel(recipient.id, "NTFY"), + ]); + ntfyEgress.error = new Error("provider timeout"); + + await expect( + dispatchNotification({ + eventType: "MEDICATION_REMINDER", + userId: record.id, + title: "record reminder", + message: "record message", + }), + ).resolves.toEqual({ + dispatched: false, + channelsAttempted: 1, + channelsSucceeded: 0, + }); + + await expect( + getPrismaClient().notificationEgressAuthorization.findFirstOrThrow({ + where: { + recordUserId: record.id, + recipientUserId: recipient.id, + channel: "NTFY", + }, + select: { eventType: true, outcome: true, completedAt: true }, + }), + ).resolves.toEqual({ + eventType: "MEDICATION_REMINDER", + outcome: "sender_threw", + completedAt: expect.any(Date), + }); + }); + + it("creates a new claim for each retry and stops after Guardian revocation", async () => { + const [record, recipient, reserveGuardian] = await Promise.all([ + createUser({ label: "record", locale: "en", managed: true }), + createUser({ label: "recipient", locale: "en" }), + createUser({ label: "reserve-guardian", locale: "de" }), + ]); + await Promise.all([ + addGuardian(record.id, recipient.id), + addGuardian(record.id, reserveGuardian.id), + addChannel(recipient.id, "NTFY"), + ]); + const grant = await getPrismaClient().accountGrant.findFirstOrThrow({ + where: { grantorId: record.id, granteeId: recipient.id }, + select: { id: true }, + }); + const payload = { + eventType: "MEDICATION_REMINDER" as const, + userId: record.id, + recordUserId: record.id, + recipientUserId: recipient.id, + title: "record reminder", + message: "record message", + }; + + ntfyEgress.error = new Error("provider timeout"); + await dispatchNotification(payload); + await getPrismaClient().notificationChannel.update({ + where: { userId_type: { userId: recipient.id, type: "NTFY" } }, + data: { consecutiveFailures: 0, nextRetryAt: null }, + }); + ntfyEgress.error = null; + ntfyEgress.outcome = { ok: false, reason: "provider_unavailable" }; + await dispatchNotification(payload); + + const beforeRevocation = + await getPrismaClient().notificationEgressAuthorization.findMany({ + where: { + recordUserId: record.id, + recipientUserId: recipient.id, + channel: "NTFY", + }, + select: { id: true, outcome: true }, + orderBy: { authorizedAt: "asc" }, + }); + expect(beforeRevocation).toEqual([ + { id: expect.any(String), outcome: "sender_threw" }, + { id: expect.any(String), outcome: "transient_failure" }, + ]); + expect(beforeRevocation[0]!.id).not.toBe(beforeRevocation[1]!.id); + + await revokeGrant({ grantId: grant.id, grantorId: record.id }); + ntfyEgress.outcome = null; + await expect(dispatchNotification(payload)).resolves.toEqual({ + dispatched: false, + channelsAttempted: 0, + channelsSucceeded: 0, + }); + expect(sent.ntfy).toEqual([]); + await expect( + getPrismaClient().notificationEgressAuthorization.count({ + where: { + recordUserId: record.id, + recipientUserId: recipient.id, + channel: "NTFY", + }, + }), + ).resolves.toBe(2); }); }); -async function expectPromiseToRemainPending(promise: Promise) { +async function expectPromiseToSettle(promise: Promise) { const settled = await Promise.race([ promise.then( () => true, () => true, ), - new Promise((resolve) => setTimeout(() => resolve(false), 25)), + new Promise((resolve) => setTimeout(() => resolve(false), 250)), ]); - expect(settled).toBe(false); + expect(settled).toBe(true); } diff --git a/tests/integration/notification-attribution.test.ts b/tests/integration/notification-attribution.test.ts index ba6d50e16..749eccbf7 100644 --- a/tests/integration/notification-attribution.test.ts +++ b/tests/integration/notification-attribution.test.ts @@ -2,8 +2,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { getPrismaClient, truncateAllTables } from "./setup"; import { + claimManagedGuardianEgressAuthorization, resolveNotificationDeliveryIdentity, - withManagedGuardianEgressAuthorization, } from "@/lib/notifications/delivery-identity"; import { claimNotificationEvent } from "@/lib/notifications/reminder-dedup"; import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record"; @@ -157,7 +157,42 @@ describe("notification attribution (real Postgres)", () => { ); }); - it("refuses an expired Guardian after identity preflight but before provider egress", async () => { + it("refuses a revoked Guardian after identity preflight but before the final egress claim", async () => { + const recordUser = await createUser("expiry-record", true); + const recipientUser = await createUser("expiry-recipient"); + const grant = await createGuardianGrant({ + recordUserId: recordUser.id, + recipientUserId: recipientUser.id, + }); + 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: { revokedAt: new Date(), revokedBy: "GRANTOR" }, + }); + + await expect( + claimManagedGuardianEgressAuthorization( + delivery!, + "NTFY", + "MEDICATION_REMINDER", + ), + ).resolves.toBeNull(); + expect( + await getPrismaClient().notificationEgressAuthorization.count(), + ).toBe(0); + expect(await getPrismaClient().pushAttempt.count()).toBe(0); + }); + + it("refuses an expired Guardian after identity preflight but before the final egress claim", async () => { const recordUser = await createUser("expiry-record", true); const recipientUser = await createUser("expiry-recipient"); const grant = await createGuardianGrant({ @@ -179,15 +214,18 @@ describe("notification attribution (real Postgres)", () => { where: { id: grant.id }, data: { expiresAt: new Date(Date.now() - 1) }, }); - let sent = false; await expect( - withManagedGuardianEgressAuthorization(delivery!, async () => { - sent = true; - return { ok: true }; - }), + claimManagedGuardianEgressAuthorization( + delivery!, + "NTFY", + "MEDICATION_REMINDER", + ), ).resolves.toBeNull(); - expect(sent).toBe(false); + expect( + await getPrismaClient().notificationEgressAuthorization.count(), + ).toBe(0); + expect(await getPrismaClient().pushAttempt.count()).toBe(0); }); it("allows exactly one concurrent claim for a notification event", async () => {