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

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");

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")
 }
 

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 () => {

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.
  */

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<T>(
+  delivery: NotificationDeliveryIdentity,
+  send: () => Promise<T>,
+): Promise<T | null> {
+  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<T>(
+  recordUserId: string,
+  recipientUserId: string,
+  operation: () => Promise<T>,
+  options?: { maxWait: number; timeout: number },
+): Promise<T | 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 guardian = await tx.accountGrant.findFirst({
+          where: {
+            grantorId: recordUserId,
+            granteeId: recipientUserId,
+            ...activeGuardianWhere(new Date()),
+          },
+          select: { id: true },
+        });
+        if (!guardian) return null;
+
+        return operation();
+      }),
+    options,
   );
 }

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({

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<void>((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<void>((resolve) => {

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();