feat: separate notification event anchors

- Store record-scoped scheduler events outside delivery attempts\n- Keep delivery diagnostics free of bookkeeping rows
This commit is contained in:
Marc-André Bombeck committed 2026-08-06 04:32:52 +02:00
1 parent 2999b75ea6
commit 9380ea0f18
8 files changed
+85 -48

No files matched your search

@@ -34,6 +34,22 @@ CREATE INDEX "push_attempts_recipient_user_id_created_at_desc_idx"
CREATE INDEX "push_attempts_record_user_id_created_at_desc_idx"
ON "push_attempts" ("record_user_id", "created_at" DESC);
CREATE TABLE "notification_events" (
"id" TEXT NOT NULL,
"record_user_id" TEXT NOT NULL,
"event_type" TEXT NOT NULL,
"dedup_key" TEXT NOT NULL,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "notification_events_pkey" PRIMARY KEY ("id"),
CONSTRAINT "notification_events_record_user_id_fkey"
FOREIGN KEY ("record_user_id") REFERENCES "users"("id")
ON DELETE CASCADE ON UPDATE CASCADE
);
CREATE INDEX "notification_events_record_user_id_event_type_dedup_key_created_at_desc_idx"
ON "notification_events" ("record_user_id", "event_type", "dedup_key", "created_at" DESC);
CREATE FUNCTION enforce_push_attempt_attribution()
RETURNS TRIGGER
LANGUAGE plpgsql
+17
View File
@@ -862,6 +862,7 @@ model User {
// `reminder-worker.ts`).
pushAttemptsForRecord PushAttempt[] @relation("PushAttemptRecordUser")
pushAttemptsReceived PushAttempt[] @relation("PushAttemptRecipientUser")
notificationEvents NotificationEvent[]
// v1.10.3 — per-(user, day) Strain personal-anchor cache (day-total
// TRIMP + running EWMA reference) so the nightly Strain engine anchors
// the 0–100 map to the user's own training load without re-integrating
@@ -5804,6 +5805,22 @@ model ConsentReceipt {
// `channel` is the dispatcher `ChannelType` string (`APNS`,
// `WEB_PUSH`, `TELEGRAM`, `NTFY`).
//
// Record-scoped notification events are durable event anchors rather than
// channel delivery records. They intentionally name no recipient, which keeps
// scheduler deduplication independent from future Guardian fan-out.
model NotificationEvent {
id String @id @default(cuid())
recordUserId String @map("record_user_id")
eventType String @map("event_type")
dedupKey String @map("dedup_key")
createdAt DateTime @default(now()) @map("created_at")
recordUser User @relation(fields: [recordUserId], references: [id], onDelete: Cascade)
@@index([recordUserId, eventType, dedupKey, createdAt(sort: Desc)])
@@map("notification_events")
}
// `result` is one of `"ok"` | `"error"` | `"skipped"` — `skipped`
// covers the "no recipient" / "config missing" soft cases.
//
@@ -68,7 +68,6 @@ vi.mock("@/lib/crypto", () => ({
import { GET } from "../route";
import { prisma } from "@/lib/db";
import { requireAdmin, HttpError } from "@/lib/api-handler";
import { REMINDER_DEDUP_CHANNEL } from "@/lib/notifications/reminder-dedup";
const ADMIN_USER_ID = "admin-1";
const ADMIN_CTX = {
@@ -405,19 +404,15 @@ describe("GET /api/admin/notifications/diagnostic — recentPushAttempts", () =>
expect(args.take).toBe(20);
});
it("excludes reminder-dedup bookkeeping rows so the last 20 are real sends", async () => {
// The medication tick anchors its per-slot dedup in this table under a
// sentinel channel. Letting those rows into the trailing window would
// push the actual delivery attempts — the thing an operator opens this
// endpoint for — off the list.
it("reads recipient-scoped delivery attempts without bookkeeping sentinels", async () => {
vi.mocked(requireAdmin).mockResolvedValue(ADMIN_CTX);
vi.mocked(prisma.pushAttempt.findMany).mockResolvedValue([] as never);
await GET();
const args = vi.mocked(prisma.pushAttempt.findMany).mock.calls[0]?.[0] as {
where: { channel?: { not?: string } };
where: { recipientUserId?: string };
};
expect(args.where.channel).toEqual({ not: REMINDER_DEDUP_CHANNEL });
expect(args.where).toEqual({ recipientUserId: ADMIN_USER_ID });
});
});
@@ -4,7 +4,6 @@ import { annotate } from "@/lib/logging/context";
import { apiSuccess } from "@/lib/api-response";
import { decrypt } from "@/lib/crypto";
import type { ChannelType } from "@/lib/notifications/types";
import { REMINDER_DEDUP_CHANNEL } from "@/lib/notifications/reminder-dedup";
export const dynamic = "force-dynamic";
@@ -176,12 +175,8 @@ export const GET = apiHandler(async () => {
// recipient index keeps the bounded diagnostic read independent of the
// account's lifetime notification volume.
prisma.pushAttempt.findMany({
// The reminder tick's per-slot dedup anchors live in this table under
// a sentinel channel. They are bookkeeping, not delivery attempts,
// and would push the real sends out of the trailing window.
where: {
recipientUserId: user.id,
channel: { not: REMINDER_DEDUP_CHANNEL },
},
orderBy: { createdAt: "desc" },
take: 20,
@@ -37,7 +37,6 @@ vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
import { GET } from "../route";
import { prisma } from "@/lib/db";
import { requireAdmin, HttpError } from "@/lib/api-handler";
import { REMINDER_DEDUP_CHANNEL } from "@/lib/notifications/reminder-dedup";
import { NextRequest } from "next/server";
const ADMIN_CTX = {
@@ -140,15 +139,12 @@ describe("GET /api/admin/notifications/health — aggregate", () => {
]);
});
it("excludes reminder-dedup bookkeeping rows from the channel picture", async () => {
// The medication tick anchors its per-slot dedup in the same table
// under a sentinel channel. Those rows are not delivery attempts, so
// counting them would misreport what the channels are doing.
it("groups every delivery attempt in the selected window", async () => {
await GET(req());
const args = vi.mocked(prisma.pushAttempt.groupBy).mock.calls[0][0] as {
where: { channel?: { not?: string } };
where: { createdAt?: { gte?: Date } };
};
expect(args.where.channel).toEqual({ not: REMINDER_DEDUP_CHANNEL });
expect(args.where.createdAt?.gte).toBeInstanceOf(Date);
});
it("defaults the window to 24h and clamps out-of-range values", async () => {
@@ -2,7 +2,6 @@ import { prisma } from "@/lib/db";
import { apiHandler, requireAdmin } from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import { apiSuccess } from "@/lib/api-response";
import { REMINDER_DEDUP_CHANNEL } from "@/lib/notifications/reminder-dedup";
export const dynamic = "force-dynamic";
@@ -59,12 +58,8 @@ export const GET = apiHandler(async (request) => {
const [grouped, disabled] = await Promise.all([
prisma.pushAttempt.groupBy({
by: ["channel", "result"],
// The reminder tick anchors its per-slot dedup in the same table
// under a sentinel channel. Those rows record a decision, not a
// delivery, so they stay out of the per-channel health picture.
where: {
createdAt: { gte: since },
channel: { not: REMINDER_DEDUP_CHANNEL },
},
_count: { _all: true },
}),
+11 -23
View File
@@ -8,9 +8,9 @@
* ntfy had no row, so the tick re-sent the same overdue notice on every
* pass for the rest of the local day.
*
* The anchor now lives in `push_attempts`, the ledger the daily briefing
* and the illness escalation already use for the same purpose. Two
* properties matter and both are deliberate:
* The anchor now lives in `notification_events`, a record-scoped event ledger
* rather than a channel delivery record. Two properties matter and both are
* deliberate:
*
* - **Written per dispatched slot, not per successful send.** Anchoring
* on a delivered push would re-open the flood the moment a channel
@@ -19,24 +19,14 @@
* for that phase rather than risking a repeat; the next phase still
* escalates, and the next local day starts clean.
*
* The rows are bookkeeping, not delivery attempts, so they carry the
* `DEDUP` channel sentinel and the operator-facing notification surfaces
* filter them out.
*/
import type { PrismaClient } from "@/generated/prisma/client";
/**
* Channel sentinel for a bookkeeping row. Never a real delivery target —
* `/api/admin/notifications/{health,diagnostic}` exclude it so the
* per-channel health picture stays a picture of actual sends.
*/
export const REMINDER_DEDUP_CHANNEL = "DEDUP";
/**
* How far back the anchor lookup reads. The key already carries the local
* date, so anything older than this cannot match; the bound exists to keep
* the read on the `(user_id, created_at DESC)` index. Two days covers every
* timezone offset plus a DST shift.
* the read on the record/event/key index. Two days covers every timezone
* offset plus a DST shift.
*/
const ANCHOR_LOOKBACK_MS = 48 * 60 * 60 * 1000;
@@ -79,11 +69,11 @@ export async function hasReminderDedupAnchor(
input: { userId: string; eventType: string; reason: string; now: Date },
): Promise<boolean> {
try {
const row = await prisma.pushAttempt.findFirst({
const row = await prisma.notificationEvent.findFirst({
where: {
userId: input.userId,
recordUserId: input.userId,
eventType: input.eventType,
reason: input.reason,
dedupKey: input.reason,
createdAt: { gte: new Date(input.now.getTime() - ANCHOR_LOOKBACK_MS) },
},
select: { id: true },
@@ -104,13 +94,11 @@ export async function writeReminderDedupAnchor(
input: { userId: string; eventType: string; reason: string },
): Promise<void> {
try {
await prisma.pushAttempt.create({
await prisma.notificationEvent.create({
data: {
userId: input.userId,
channel: REMINDER_DEDUP_CHANNEL,
recordUserId: input.userId,
eventType: input.eventType,
result: "skipped",
reason: input.reason,
dedupKey: input.reason,
},
});
} catch {
@@ -2,6 +2,10 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
import { getPrismaClient, truncateAllTables } from "./setup";
import { resolveNotificationDeliveryIdentity } from "@/lib/notifications/delivery-identity";
import {
hasReminderDedupAnchor,
writeReminderDedupAnchor,
} from "@/lib/notifications/reminder-dedup";
import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record";
let sequence = 0;
@@ -104,6 +108,37 @@ describe("notification attribution (real Postgres)", () => {
});
});
it("persists a managed reminder anchor as a record event without a delivery attempt", async () => {
const recordUser = await createUser("managed-record", true);
const client = getPrismaClient();
const now = new Date();
await writeReminderDedupAnchor(client, {
userId: recordUser.id,
eventType: "MEDICATION_REMINDER",
reason: "med:slot",
});
await expect(
hasReminderDedupAnchor(client, {
userId: recordUser.id,
eventType: "MEDICATION_REMINDER",
reason: "med:slot",
now,
}),
).resolves.toBe(true);
expect(await client.pushAttempt.count()).toBe(0);
const event = await client.notificationEvent.findFirstOrThrow({
where: { recordUserId: recordUser.id, dedupKey: "med:slot" },
select: { id: true },
});
await client.user.delete({ where: { id: recordUser.id } });
expect(
await client.notificationEvent.findUnique({ where: { id: event.id } }),
).toBeNull();
});
it("persistence maps an omitted pair from the legacy recipient", async () => {
const self = await createUser("legacy");