Claim reminder slots before egress, classify the bulk race, fold the vault probes

- The measurement-reminder tick claims its slot in the record-event
  ledger BEFORE any provider egress (the documented claim-first pattern
  from reminder-dedup.ts); a crash between dispatch and the nextDueAt
  advance can no longer double-send.
- mood-entries/bulk classifies a lost P2002 race as the duplicate it is
  (with the winning row id) instead of a skip carrying a raw Prisma
  message — the row IS stored, the racing twin wrote it. The workouts
  batch already re-derives per-entry status from RETURNING.
- The vault list's content-index and thumbnail probes ride the same
  Promise.all as the link loads: four independent grouped queries, no
  serialized round-trips on a 4s-polled list.
This commit is contained in:
Marc-André Bombeck committed 2026-08-14 06:06:43 +02:00
1 parent 14e9edfa7a
commit ac44b35867
5 files changed
+208 -36

No files matched your search

+28 -31
View File
@@ -631,40 +631,37 @@ export const GET = apiHandler(async (request: Request) => {
}
}
// Condition links for the page in ONE grouped query (no N+1).
const [linkMap, visitLinkMap] = await Promise.all([
loadConditionLinks(
user.id,
page.map((d) => d.id),
),
loadDocumentEncounterLinks(
user.id,
page.map((d) => d.id),
),
// Condition/visit links + the content-index and thumbnail probes for the
// page, all in ONE Promise.all — four independent grouped queries (no
// N+1, never a ciphertext/blob column), and the vault polls this list
// every few seconds so two serialized round-trips were pure added
// latency (A5-4).
const pageIds = page.map((d) => d.id);
const [linkMap, visitLinkMap, indexedRows, thumbRows] = await Promise.all([
loadConditionLinks(user.id, pageIds),
loadDocumentEncounterLinks(user.id, pageIds),
// Which of the page's documents have a content index (drives the
// searchable status + the provenance the UI reads to tell an AI-read
// document from a locally-indexed one).
page.length > 0
? prisma.documentContentIndex.findMany({
where: { userId: user.id, documentId: { in: pageIds } },
select: { documentId: true, source: true },
})
: Promise.resolve([]),
// Which of the page's documents have a preview thumbnail (gates the
// card's <img>). Grouped on the 1:1 side table.
page.length > 0
? prisma.documentThumbnail.findMany({
where: { userId: user.id, documentId: { in: pageIds } },
select: { documentId: true },
})
: Promise.resolve([]),
]);
// Which of the page's documents have a content index (drives the searchable
// status + the provenance the UI reads to tell an AI-read document from a
// locally-indexed one). One grouped query; never the ciphertext.
const indexSources = new Map<string, string>();
if (page.length > 0) {
const indexed = await prisma.documentContentIndex.findMany({
where: { userId: user.id, documentId: { in: page.map((d) => d.id) } },
select: { documentId: true, source: true },
});
for (const row of indexed) indexSources.set(row.documentId, row.source);
}
// Which of the page's documents have a preview thumbnail (gates the card's
// <img>). One grouped query on the 1:1 side table; never the blob column.
for (const row of indexedRows) indexSources.set(row.documentId, row.source);
const thumbnailIds = new Set<string>();
if (page.length > 0) {
const thumbs = await prisma.documentThumbnail.findMany({
where: { userId: user.id, documentId: { in: page.map((d) => d.id) } },
select: { documentId: true },
});
for (const row of thumbs) thumbnailIds.add(row.documentId);
}
for (const row of thumbRows) thumbnailIds.add(row.documentId);
annotate({
action: { name: "documents.inbound.list" },
@@ -19,7 +19,12 @@ import { NextRequest } from "next/server";
// assertion below is vacuous. It SHARES the model mocks so every existing
// per-model assertion in this file keeps working unchanged.
const { moodEntryMock, txClient } = vi.hoisted(() => {
const moodEntryMock = { findMany: vi.fn(), upsert: vi.fn() };
const moodEntryMock = {
findMany: vi.fn(),
upsert: vi.fn(),
// v1.37.19 (A6-18) — the P2002-race arm resolves the winning row.
findFirst: vi.fn(),
};
return {
moodEntryMock,
txClient: { __brand: "transaction-client", moodEntry: moodEntryMock },
@@ -462,6 +467,49 @@ describe("POST /api/mood-entries/bulk — structured tagKeys (v1.12.0)", () => {
expect(json.data.entries[0].status).toBe("skipped");
expect(json.data.entries[1].status).toBe("inserted");
});
// Watched red: with the P2002 arm removed from the loop catch (the
// pre-v1.37.19 route), this fails — a concurrent batch losing the
// `(userId, source, externalId)` race was reported as "skipped" with a
// raw Prisma message although the row IS stored (the racing twin wrote
// it). It is the plain duplicate the sequential probe would have found.
it("classifies a lost P2002 race as duplicate with the winning row id", async () => {
const p2002 = Object.assign(new Error("Unique constraint failed"), {
code: "P2002",
});
vi.mocked(moodEntryMock.upsert).mockRejectedValueOnce(p2002);
vi.mocked(moodEntryMock.findFirst).mockResolvedValueOnce({
id: "winner-1",
} as never);
vi.mocked(moodEntryMock.findMany).mockResolvedValueOnce([] as never);
const res = await POST(
postReq({
entries: [
{
mood: "GUT",
moodLoggedAt: "2026-05-16T08:00:00.000Z",
source: "MOODLOG",
externalId: "hk-race-1",
},
],
}),
);
expect(res.status).toBe(200);
const json = (await res.json()) as {
data: {
inserted: number;
duplicates: number;
skipped: unknown[];
entries: Array<{ status: string; id?: string; externalId?: string }>;
};
};
expect(json.data.entries[0].status).toBe("duplicate");
expect(json.data.entries[0].id).toBe("winner-1");
expect(json.data.duplicates).toBe(1);
expect(json.data.inserted).toBe(0);
expect(json.data.skipped).toEqual([]);
});
});
describe("POST /api/mood-entries/bulk — unstable external ids", () => {
+33
View File
@@ -522,6 +522,39 @@ async function postBulk(request: NextRequest): Promise<Response> {
});
}
} catch (err: unknown) {
// v1.37.19 (A6-18) — P2002 = a concurrent batch raced the same
// NULL-distinct `(userId, source, externalId)` key and won. That is
// the plain duplicate the sequential probe would have classified;
// reporting it as "skipped" with a raw Prisma message told the
// client its row was dropped when it is in fact stored. Resolve the
// winning row's id so the caller's cursor advances, exactly like
// the intake bulk route.
if (
typeof err === "object" &&
err !== null &&
"code" in err &&
(err as { code: string }).code === "P2002" &&
entry.externalId
) {
const winner = await prisma.moodEntry.findFirst({
where: {
userId: user.id,
// Same resolution as the upsert key: the schema defaults
// `source` to "MANUAL" at parse time.
source: entry.source,
externalId: entry.externalId,
},
select: { id: true },
});
duplicates += 1;
results.push({
index: i,
status: "duplicate",
...(winner ? { id: winner.id } : {}),
externalId: entry.externalId,
});
continue;
}
const reason =
err instanceof Error ? err.message.slice(0, 120) : "upsert_failed";
skipped.push({ index: i, reason });
@@ -16,7 +16,20 @@
*
* The Prisma surface is stubbed manually to avoid a testcontainer boot.
*/
import { describe, it, expect, vi } from "vitest";
import { beforeEach, describe, it, expect, vi } from "vitest";
// v1.37.19 (A6-7) — the tick claims the slot BEFORE provider egress via
// the shared record-event ledger. The manual Prisma double has no
// notification_events surface, so the claim is mocked controllable:
// default granted, individual cases flip it to prove the guard.
const claimState = vi.hoisted(() => ({ granted: true }));
const claimMock = vi.hoisted(() => vi.fn(async () => claimState.granted));
vi.mock("@/lib/notifications/reminder-dedup", async () => {
const actual = await vi.importActual<
typeof import("@/lib/notifications/reminder-dedup")
>("@/lib/notifications/reminder-dedup");
return { ...actual, claimNotificationEvent: claimMock };
});
import {
evaluateMeasurementReminderDue,
@@ -35,6 +48,11 @@ const OK: DispatchOutcome = {
const TZ = "Europe/Berlin";
beforeEach(() => {
claimState.granted = true;
claimMock.mockClear();
});
// 09:00 Berlin in June = 07:00Z.
const NINE_LOCAL = new Date("2026-06-15T07:00:00Z");
@@ -228,6 +246,51 @@ describe("runMeasurementReminderTick", () => {
expect(updates[0].data.nextDueAt).toBeInstanceOf(Date);
});
// Watched red: with the claim-before-dispatch guard removed from the
// tick (the pre-v1.37.19 order: dispatch first, advance after), the
// denied-claim case below fails — the provider was contacted although
// another worker (or a crash-recovery replay) already owned the slot.
it("claims the slot BEFORE provider egress and skips when denied", async () => {
claimState.granted = false;
const { prisma, updates } = makePrisma({
reminders: [reminder({})],
measurementMatch: null,
});
const dispatch = vi.fn<DispatchFn>(async () => OK);
const summary = await runMeasurementReminderTick(
prisma as never,
NINE_LOCAL,
{ dispatch },
);
expect(claimMock).toHaveBeenCalledTimes(1);
expect(dispatch).not.toHaveBeenCalled();
expect(summary.skippedAlreadyClaimed).toBe(1);
expect(summary.dispatched).toBe(0);
// No advance either — the claimant owns the slot's lifecycle.
expect(updates).toHaveLength(0);
});
it("keys the claim on the reminder + the user-local date", async () => {
const { prisma } = makePrisma({
reminders: [reminder({})],
measurementMatch: null,
});
const dispatch = vi.fn<DispatchFn>(async () => OK);
await runMeasurementReminderTick(prisma as never, NINE_LOCAL, { dispatch });
expect(claimMock).toHaveBeenCalledWith(
prisma,
expect.objectContaining({
eventType: "MEASUREMENT_REMINDER",
dedupKey: expect.stringMatching(/^measurement:.+:2026-06-15$/),
}),
);
expect(dispatch).toHaveBeenCalledTimes(1);
});
it("dispatches a due reminder and advances nextDueAt past now (ledger-free dedup)", async () => {
const { prisma, updates } = makePrisma({
reminders: [reminder({})],
+34 -3
View File
@@ -34,6 +34,10 @@ import type { Locale } from "@/lib/i18n/config";
import { defaultLocale, locales } from "@/lib/i18n/config";
import { wallClockInTz } from "@/lib/tz/wall-clock";
import { dispatchNotification } from "@/lib/notifications/dispatcher";
import {
claimNotificationEvent,
REMINDER_DEDUP_LOOKBACK_MS,
} from "@/lib/notifications/reminder-dedup";
import { getEvent } from "@/lib/logging/context";
import { isModuleEnabled } from "@/lib/modules/gate";
import type { ModuleKey } from "@/lib/modules/registry";
@@ -92,6 +96,8 @@ export interface MeasurementReminderSummary {
skippedOutsideWindow: number;
skippedModuleDisabled: number;
skippedNoChannel: number;
/** v1.37.19 (A6-7) — slot already claimed for this local day (racing worker or crash-recovery replay). */
skippedAlreadyClaimed: number;
/** v1.18.1 — expired COACH course-window reminders soft-deleted this tick. */
expiredCleaned: number;
failed: number;
@@ -204,6 +210,7 @@ export async function runMeasurementReminderTick(
skippedOutsideWindow: 0,
skippedModuleDisabled: 0,
skippedNoChannel: 0,
skippedAlreadyClaimed: 0,
expiredCleaned: 0,
failed: 0,
};
@@ -325,6 +332,28 @@ export async function runMeasurementReminderTick(
// user was filed as handled and vanished from the card and the digest.
// Now, when nothing delivers, the no-channel branch below leaves the
// reminder overdue and visible.
// v1.37.19 (A6-7) — claim the slot BEFORE provider egress, the
// documented claim-first pattern from `reminder-dedup.ts`. The old
// order (dispatch, then advance `nextDueAt`) double-sent when the
// worker crashed between the two; the claim is the durable record
// that this reminder's due cycle went out today. The trade is the
// same one the medication tick accepts: a crash mid-dispatch burns
// the slot for this local day rather than risking a repeat, and the
// next local day starts clean.
const localDate = new Date(now).toLocaleDateString("sv-SE", {
timeZone: timezone,
});
const claimed = await claimNotificationEvent(prisma, {
recordUserId: reminder.user.id,
eventType: "MEASUREMENT_REMINDER",
dedupKey: `measurement:${reminder.id}:${localDate}`,
since: new Date(now.getTime() - REMINDER_DEDUP_LOOKBACK_MS),
});
if (!claimed) {
summary.skippedAlreadyClaimed += 1;
continue;
}
const { title, body } = buildMeasurementReminderPayload(
reminder.user.locale,
reminder.label,
@@ -359,9 +388,11 @@ export async function runMeasurementReminderTick(
},
});
// No channel succeeded — leave `nextDueAt` where it is so the next
// tick (or the user's next channel) retries. The reminder simply
// stays overdue, which the surface already shows as "überfällig".
// No channel succeeded — leave `nextDueAt` where it is. The claim
// above holds for the rest of the local day (claim-first burns the
// slot rather than re-flooding a failing channel — see
// reminder-dedup.ts); the reminder stays overdue and visible on the
// surface, and the next local day retries with a fresh key.
if (!outcome.dispatched) {
summary.skippedNoChannel += 1;
continue;