test: cover atomic notification event claims

- race record event claims through PostgreSQL\n- require safety producers to honor the claim

src/lib/illness/__tests__/red-flag-notify.test.ts

@@ -11,6 +11,7 @@ const findFirstMock = vi.fn();
 const createMock = vi.fn();
 const dispatchMock = vi.fn();
 const userFindUniqueMock = vi.fn();
+const claimMock = vi.fn();
 
 vi.mock("@/lib/db", () => ({
   prisma: {
@@ -32,6 +33,10 @@ vi.mock("@/lib/notifications/dispatch-localised", () => ({
   dispatchLocalisedNotification: (...a: unknown[]) => dispatchMock(...a),
 }));
 
+vi.mock("@/lib/notifications/reminder-dedup", () => ({
+  claimNotificationEvent: (...a: unknown[]) => claimMock(...a),
+}));
+
 import { notifyIllnessRedFlag } from "../red-flag-notify";
 import type { IllnessRedFlag } from "../correlation";
 
@@ -53,6 +58,7 @@ beforeEach(() => {
   findFirstMock.mockReset().mockResolvedValue(null);
   createMock.mockReset().mockResolvedValue({});
   dispatchMock.mockReset().mockResolvedValue(undefined);
+  claimMock.mockReset().mockResolvedValue(true);
 });
 
 afterEach(() => vi.clearAllMocks());
@@ -90,29 +96,31 @@ describe("notifyIllnessRedFlag", () => {
     );
   });
 
-  it("de-dupes when a recent ledger row exists for the episode", async () => {
-    findFirstMock.mockResolvedValueOnce({ id: "prior" });
+  it("does not dispatch when a concurrent claimant already owns the episode", async () => {
+    claimMock.mockResolvedValueOnce(false);
     await notifyIllnessRedFlag({
       userId: "u1",
       episodeId: "e1",
       redFlags: [feverFlag],
     });
     expect(dispatchMock).not.toHaveBeenCalled();
-    expect(createMock).not.toHaveBeenCalled();
+    expect(claimMock).toHaveBeenCalledTimes(1);
   });
 
-  it("stamps the record event anchor before dispatching", async () => {
+  it("claims the record event anchor before dispatching", async () => {
     await notifyIllnessRedFlag({
       userId: "u1",
       episodeId: "e9",
       redFlags: [feverFlag],
     });
-    expect(createMock).toHaveBeenCalledTimes(1);
-    expect(createMock.mock.calls[0][0].data).toMatchObject({
-      recordUserId: "u1",
-      eventType: "SYSTEM_ALERT",
-      dedupKey: expect.stringContaining("e9"),
-    });
+    expect(claimMock).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({
+        recordUserId: "u1",
+        eventType: "SYSTEM_ALERT",
+        dedupKey: expect.stringContaining("e9"),
+      }),
+    );
   });
 
   it("does not dispatch or write an anchor for a managed profile", async () => {
@@ -126,8 +134,7 @@ describe("notifyIllnessRedFlag", () => {
       redFlags: [feverFlag],
     });
 
-    expect(findFirstMock).not.toHaveBeenCalled();
-    expect(createMock).not.toHaveBeenCalled();
+    expect(claimMock).not.toHaveBeenCalled();
     expect(dispatchMock).not.toHaveBeenCalled();
   });
 

src/lib/illness/__tests__/safety-floor-notify.test.ts

@@ -12,6 +12,7 @@ const findFirstMock = vi.fn();
 const createMock = vi.fn();
 const dispatchMock = vi.fn();
 const userFindUniqueMock = vi.fn();
+const claimMock = vi.fn();
 
 vi.mock("@/lib/db", () => ({
   prisma: {
@@ -33,6 +34,10 @@ vi.mock("@/lib/notifications/dispatch-localised", () => ({
   dispatchLocalisedNotification: (...a: unknown[]) => dispatchMock(...a),
 }));
 
+vi.mock("@/lib/notifications/reminder-dedup", () => ({
+  claimNotificationEvent: (...a: unknown[]) => claimMock(...a),
+}));
+
 import { notifySafetyFloor } from "../safety-floor-notify";
 import type { SafetyFloorDecision } from "../safety-floors";
 
@@ -58,6 +63,7 @@ beforeEach(() => {
   findFirstMock.mockReset().mockResolvedValue(null);
   createMock.mockReset().mockResolvedValue({});
   dispatchMock.mockReset().mockResolvedValue(undefined);
+  claimMock.mockReset().mockResolvedValue(true);
 });
 
 afterEach(() => vi.clearAllMocks());
@@ -99,21 +105,23 @@ describe("notifySafetyFloor", () => {
     expect(opts.params).toEqual({ value: 2.7, unit: "mmol/L", threshold: 3 });
   });
 
-  it("de-dupes by reason when a recent ledger row exists", async () => {
-    findFirstMock.mockResolvedValueOnce({ id: "prior" });
+  it("does not dispatch when a concurrent claimant owns the safety event", async () => {
+    claimMock.mockResolvedValueOnce(false);
     await notifySafetyFloor({ userId: "u1", decision: bpHigh });
     expect(dispatchMock).not.toHaveBeenCalled();
-    expect(createMock).not.toHaveBeenCalled();
+    expect(claimMock).toHaveBeenCalledTimes(1);
   });
 
-  it("stamps the record event anchor (keyed by reason) before dispatching", async () => {
+  it("claims the record event anchor (keyed by reason) before dispatching", async () => {
     await notifySafetyFloor({ userId: "u9", decision: bpHigh });
-    expect(createMock).toHaveBeenCalledTimes(1);
-    expect(createMock.mock.calls[0][0].data).toMatchObject({
-      recordUserId: "u9",
-      eventType: "SYSTEM_ALERT",
-      dedupKey: expect.stringContaining("bp_hypertensive"),
-    });
+    expect(claimMock).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({
+        recordUserId: "u9",
+        eventType: "SYSTEM_ALERT",
+        dedupKey: expect.stringContaining("bp_hypertensive"),
+      }),
+    );
   });
 
   it("does not dispatch or write an anchor for a managed profile", async () => {
@@ -123,8 +131,7 @@ describe("notifySafetyFloor", () => {
 
     await notifySafetyFloor({ userId: "managed-record", decision: bpHigh });
 
-    expect(findFirstMock).not.toHaveBeenCalled();
-    expect(createMock).not.toHaveBeenCalled();
+    expect(claimMock).not.toHaveBeenCalled();
     expect(dispatchMock).not.toHaveBeenCalled();
   });
 

src/lib/notifications/__tests__/reminder-dedup.test.ts

@@ -4,9 +4,13 @@ import { prisma } from "@/lib/db";
 
 const findFirstMock = vi.fn();
 const createMock = vi.fn();
+const queryRawMock = vi.fn();
+const transactionMock = vi.fn();
 
 vi.mock("@/lib/db", () => ({
   prisma: {
+    $queryRaw: (...args: unknown[]) => queryRawMock(...args),
+    $transaction: (...args: unknown[]) => transactionMock(...args),
     notificationEvent: {
       findFirst: (...args: unknown[]) => findFirstMock(...args),
       create: (...args: unknown[]) => createMock(...args),
@@ -14,29 +18,29 @@ vi.mock("@/lib/db", () => ({
   },
 }));
 
-import {
-  hasReminderDedupAnchor,
-  writeReminderDedupAnchor,
-} from "../reminder-dedup";
+import { claimNotificationEvent } from "../reminder-dedup";
 
 beforeEach(() => {
   findFirstMock.mockReset().mockResolvedValue(null);
   createMock.mockReset().mockResolvedValue({ id: "event" });
+  queryRawMock.mockReset().mockResolvedValue([{ locked: 1 }]);
+  transactionMock.mockReset().mockImplementation(async (run) => run(prisma));
 });
 
 describe("reminder dedup anchors", () => {
-  it("looks up a record-scoped event rather than a delivery attempt", async () => {
+  it("claims one record-scoped event under a short transaction lock", async () => {
     const now = new Date("2026-08-06T08:00:00.000Z");
 
     await expect(
-      hasReminderDedupAnchor(prisma as never, {
-        userId: "managed-record",
+      claimNotificationEvent(prisma as never, {
+        recordUserId: "managed-record",
         eventType: "MEDICATION_REMINDER",
-        reason: "med:slot",
-        now,
+        dedupKey: "med:slot",
+        since: now,
       }),
-    ).resolves.toBe(false);
+    ).resolves.toBe(true);
 
+    expect(queryRawMock).toHaveBeenCalledTimes(1);
     expect(findFirstMock).toHaveBeenCalledWith(
       expect.objectContaining({
         where: expect.objectContaining({
@@ -46,15 +50,6 @@ describe("reminder dedup anchors", () => {
         }),
       }),
     );
-  });
-
-  it("writes an append-only record event without inventing a recipient", async () => {
-    await writeReminderDedupAnchor(prisma as never, {
-      userId: "managed-record",
-      eventType: "MEDICATION_REMINDER",
-      reason: "med:slot",
-    });
-
     expect(createMock).toHaveBeenCalledWith({
       data: {
         recordUserId: "managed-record",
@@ -63,4 +58,19 @@ describe("reminder dedup anchors", () => {
       },
     });
   });
+
+  it("does not claim when the locked window already contains the event", async () => {
+    findFirstMock.mockResolvedValueOnce({ id: "prior" });
+
+    await expect(
+      claimNotificationEvent(prisma as never, {
+        recordUserId: "managed-record",
+        eventType: "MEDICATION_REMINDER",
+        dedupKey: "med:slot",
+        since: new Date("2026-08-06T08:00:00.000Z"),
+      }),
+    ).resolves.toBe(false);
+
+    expect(createMock).not.toHaveBeenCalled();
+  });
 });

tests/integration/notification-attribution.test.ts

@@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 import { getPrismaClient, truncateAllTables } from "./setup";
 import { resolveNotificationDeliveryIdentity } from "@/lib/notifications/delivery-identity";
 import {
+  claimNotificationEvent,
   hasReminderDedupAnchor,
   writeReminderDedupAnchor,
 } from "@/lib/notifications/reminder-dedup";
@@ -140,6 +141,79 @@ describe("notification attribution (real Postgres)", () => {
     ).toBeNull();
   });
 
+  it("allows exactly one concurrent claim for a notification event", async () => {
+    const recordUser = await createUser("claim-record", true);
+    const client = getPrismaClient();
+    const since = new Date(Date.now() - 60_000);
+    const input = {
+      recordUserId: recordUser.id,
+      eventType: "MEDICATION_REMINDER",
+      dedupKey: "med:race",
+      since,
+    };
+
+    const results = await Promise.all([
+      claimNotificationEvent(client, input),
+      claimNotificationEvent(client, input),
+    ]);
+
+    expect(results.filter(Boolean)).toHaveLength(1);
+    expect(
+      await client.notificationEvent.count({
+        where: {
+          recordUserId: recordUser.id,
+          eventType: "MEDICATION_REMINDER",
+          dedupKey: "med:race",
+        },
+      }),
+    ).toBe(1);
+  });
+
+  it("uses the rolling window and allows a retry after a failed claim", async () => {
+    const client = getPrismaClient();
+    const missingRecordId = `missing-claim-${sequence++}`;
+    const missingInput = {
+      recordUserId: missingRecordId,
+      eventType: "SYSTEM_ALERT",
+      dedupKey: "safety:retry",
+      since: new Date(Date.now() - 24 * 60 * 60 * 1000),
+    };
+
+    await expect(claimNotificationEvent(client, missingInput)).resolves.toBe(
+      false,
+    );
+    await client.user.create({
+      data: {
+        id: missingRecordId,
+        username: `claim-retry-${sequence++}`,
+        email: `claim-retry-${sequence++}@example.test`,
+      },
+    });
+    await expect(claimNotificationEvent(client, missingInput)).resolves.toBe(
+      true,
+    );
+
+    const recordUser = await createUser("expired-claim", true);
+    const now = new Date();
+    await client.notificationEvent.create({
+      data: {
+        recordUserId: recordUser.id,
+        eventType: "SYSTEM_ALERT",
+        dedupKey: "safety:window",
+        createdAt: new Date(now.getTime() - 25 * 60 * 60 * 1000),
+      },
+    });
+
+    await expect(
+      claimNotificationEvent(client, {
+        recordUserId: recordUser.id,
+        eventType: "SYSTEM_ALERT",
+        dedupKey: "safety:window",
+        since: new Date(now.getTime() - 24 * 60 * 60 * 1000),
+      }),
+    ).resolves.toBe(true);
+  });
+
   it("keeps Telegram reminder tracking independent for each recipient slot", async () => {
     const recordUser = await createUser("telegram-record", true);
     const firstRecipient = await createUser("telegram-recipient-a");