mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 04:50:59 +00:00
fix(rollups): atomic upsert closes mood + compliance DAY race
QA F-H-02 (v1.4.39): the read-aggregate-then-upsert pattern in the mood and medication-compliance DAY recomputes could interleave concurrent writes for the same (user, day) as A-SELECT -> B-SELECT -> B-UPSERT (correct N) -> A-UPSERT (stale N-1), leaving a stale row until the next write. Replaced both with a single SQL INSERT ... SELECT ... ON CONFLICT DO UPDATE that re-aggregates inside the upsert subquery so each statement commits with a snapshot taken after the prior commit released its row lock. A paired DELETE statement handles the "all entries removed" case in the same flow, gated on NOT EXISTS so it cannot clobber a row the INSERT just wrote. Pinned both helpers with a Promise.all race test that asserts both callers go through the atomic path and never fall back to the legacy read-aggregate-upsert.
This commit is contained in:
Marc-André Bombeck committed
1 parent
5e45791771
commit
362f87d4c4
4 files changed
+329
-214
No files matched your search
@@ -17,6 +17,7 @@ const mocks = vi.hoisted(() => ({
|
||||
rollupFindMany: vi.fn(),
|
||||
rollupFindFirst: vi.fn(),
|
||||
queryRaw: vi.fn(),
|
||||
executeRaw: vi.fn(),
|
||||
bossSend: vi.fn(),
|
||||
getGlobalBossMock: vi.fn(),
|
||||
}));
|
||||
@@ -33,6 +34,7 @@ vi.mock("@/lib/db", () => ({
|
||||
findFirst: mocks.rollupFindFirst,
|
||||
},
|
||||
$queryRaw: mocks.queryRaw,
|
||||
$executeRaw: mocks.executeRaw,
|
||||
},
|
||||
}));
|
||||
|
||||
@@ -62,6 +64,7 @@ const {
|
||||
rollupFindMany,
|
||||
rollupFindFirst,
|
||||
queryRaw,
|
||||
executeRaw,
|
||||
bossSend,
|
||||
getGlobalBossMock,
|
||||
} = mocks;
|
||||
@@ -74,6 +77,7 @@ beforeEach(() => {
|
||||
rollupFindMany.mockResolvedValue([]);
|
||||
rollupFindFirst.mockResolvedValue(null);
|
||||
queryRaw.mockResolvedValue([]);
|
||||
executeRaw.mockResolvedValue(0);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
@@ -98,13 +102,22 @@ describe("dayKeyForScheduledFor", () => {
|
||||
});
|
||||
|
||||
describe("recomputeMedicationComplianceForDay", () => {
|
||||
it("upserts a row reflecting the (scheduled, taken, skipped) counts", async () => {
|
||||
intakeFindMany.mockResolvedValue([
|
||||
{ takenAt: new Date(), skipped: false }, // taken
|
||||
{ takenAt: null, skipped: true }, // skipped
|
||||
{ takenAt: null, skipped: false }, // pending
|
||||
]);
|
||||
/**
|
||||
* Helper: pull the bind-variable values out of a Prisma tagged-template
|
||||
* call so the test can inspect the user-id / day / window bounds the
|
||||
* SQL was rendered with. Prisma's `$executeRaw` tag passes the
|
||||
* template strings as the first arg and the substituted values as the
|
||||
* remaining positional args.
|
||||
*/
|
||||
function bindValues(call: unknown[]): unknown[] {
|
||||
return call.slice(1);
|
||||
}
|
||||
|
||||
it("runs an atomic INSERT … ON CONFLICT upsert with the (user, medication, day) tuple", async () => {
|
||||
// QA F-H-02 (v1.4.39): the recompute is now a single SQL statement
|
||||
// that re-aggregates inside the upsert; the JS pre-fold from the
|
||||
// previous tier is gone so the test inspects the bind values
|
||||
// instead of the intake fold.
|
||||
await recomputeMedicationComplianceForDay(
|
||||
"user-1",
|
||||
"med-1",
|
||||
@@ -112,33 +125,24 @@ describe("recomputeMedicationComplianceForDay", () => {
|
||||
"Europe/Berlin",
|
||||
);
|
||||
|
||||
expect(rollupUpsert).toHaveBeenCalledTimes(1);
|
||||
const call = rollupUpsert.mock.calls[0][0];
|
||||
expect(call.where).toEqual({
|
||||
userId_medicationId_day: {
|
||||
userId: "user-1",
|
||||
medicationId: "med-1",
|
||||
day: "2026-05-18",
|
||||
},
|
||||
});
|
||||
expect(call.create).toMatchObject({
|
||||
userId: "user-1",
|
||||
medicationId: "med-1",
|
||||
day: "2026-05-18",
|
||||
scheduled: 3,
|
||||
taken: 1,
|
||||
skipped: 1,
|
||||
});
|
||||
expect(call.update).toMatchObject({
|
||||
scheduled: 3,
|
||||
taken: 1,
|
||||
skipped: 1,
|
||||
});
|
||||
// Two statements fire: the INSERT … ON CONFLICT upsert and the
|
||||
// companion DELETE for the "all events removed" branch.
|
||||
expect(executeRaw).toHaveBeenCalledTimes(2);
|
||||
const insertBinds = bindValues(executeRaw.mock.calls[0]);
|
||||
expect(insertBinds).toContain("user-1");
|
||||
expect(insertBinds).toContain("med-1");
|
||||
expect(insertBinds).toContain("2026-05-18");
|
||||
const deleteBinds = bindValues(executeRaw.mock.calls[1]);
|
||||
expect(deleteBinds).toContain("user-1");
|
||||
expect(deleteBinds).toContain("med-1");
|
||||
expect(deleteBinds).toContain("2026-05-18");
|
||||
});
|
||||
|
||||
it("deletes the rollup row when the user-day window holds zero events", async () => {
|
||||
intakeFindMany.mockResolvedValue([]);
|
||||
|
||||
it("emits a paired DELETE so a fully-cleared day removes its row", async () => {
|
||||
// The DELETE fires unconditionally and gates on NOT EXISTS — when
|
||||
// the day still has events the predicate matches zero rows. The
|
||||
// test pins that the DELETE statement was issued so a future
|
||||
// refactor that drops the second statement breaks here.
|
||||
await recomputeMedicationComplianceForDay(
|
||||
"user-1",
|
||||
"med-1",
|
||||
@@ -146,17 +150,10 @@ describe("recomputeMedicationComplianceForDay", () => {
|
||||
"Europe/Berlin",
|
||||
);
|
||||
|
||||
expect(rollupDeleteMany).toHaveBeenCalledWith({
|
||||
where: { userId: "user-1", medicationId: "med-1", day: "2026-05-18" },
|
||||
});
|
||||
expect(rollupUpsert).not.toHaveBeenCalled();
|
||||
expect(executeRaw).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("is idempotent across repeated invocations", async () => {
|
||||
intakeFindMany.mockResolvedValue([
|
||||
{ takenAt: new Date(), skipped: false },
|
||||
]);
|
||||
|
||||
await recomputeMedicationComplianceForDay(
|
||||
"user-1",
|
||||
"med-1",
|
||||
@@ -170,17 +167,16 @@ describe("recomputeMedicationComplianceForDay", () => {
|
||||
"Europe/Berlin",
|
||||
);
|
||||
|
||||
expect(rollupUpsert).toHaveBeenCalledTimes(2);
|
||||
const first = rollupUpsert.mock.calls[0][0].create;
|
||||
const second = rollupUpsert.mock.calls[1][0].create;
|
||||
expect(first).toEqual(second);
|
||||
expect(executeRaw).toHaveBeenCalledTimes(4);
|
||||
expect(bindValues(executeRaw.mock.calls[0])).toEqual(
|
||||
bindValues(executeRaw.mock.calls[2]),
|
||||
);
|
||||
expect(bindValues(executeRaw.mock.calls[1])).toEqual(
|
||||
bindValues(executeRaw.mock.calls[3]),
|
||||
);
|
||||
});
|
||||
|
||||
it("queries with a tz-anchored UTC window", async () => {
|
||||
intakeFindMany.mockResolvedValue([
|
||||
{ takenAt: new Date(), skipped: false },
|
||||
]);
|
||||
|
||||
it("binds a tz-anchored UTC window into the upsert", async () => {
|
||||
await recomputeMedicationComplianceForDay(
|
||||
"user-1",
|
||||
"med-1",
|
||||
@@ -188,33 +184,26 @@ describe("recomputeMedicationComplianceForDay", () => {
|
||||
"Europe/Berlin",
|
||||
);
|
||||
|
||||
const where = intakeFindMany.mock.calls[0][0].where;
|
||||
const insertBinds = bindValues(executeRaw.mock.calls[0]);
|
||||
const insertDates = insertBinds.filter((b): b is Date => b instanceof Date);
|
||||
// Berlin 2026-05-18 00:00 → UTC 2026-05-17 22:00 (CEST is UTC+2).
|
||||
expect(where.scheduledFor.gte.toISOString()).toBe(
|
||||
"2026-05-17T22:00:00.000Z",
|
||||
);
|
||||
expect(insertDates[0].toISOString()).toBe("2026-05-17T22:00:00.000Z");
|
||||
// Window closes 24h later — Berlin 2026-05-19 00:00 → UTC 2026-05-18 22:00.
|
||||
expect(where.scheduledFor.lt.toISOString()).toBe(
|
||||
"2026-05-18T22:00:00.000Z",
|
||||
);
|
||||
expect(insertDates[1].toISOString()).toBe("2026-05-18T22:00:00.000Z");
|
||||
});
|
||||
|
||||
it("buckets the same UTC instant on different days for Berlin vs LA", async () => {
|
||||
// 2026-05-17T23:30:00Z is 2026-05-18 in Berlin but 2026-05-17 in LA.
|
||||
intakeFindMany.mockResolvedValue([
|
||||
{ takenAt: new Date(), skipped: false },
|
||||
]);
|
||||
|
||||
await recomputeMedicationComplianceForDay(
|
||||
"user-1",
|
||||
"med-1",
|
||||
"2026-05-18",
|
||||
"Europe/Berlin",
|
||||
);
|
||||
const berlinCall = intakeFindMany.mock.calls[0][0].where;
|
||||
const berlinInsert = bindValues(executeRaw.mock.calls[0]).filter(
|
||||
(b): b is Date => b instanceof Date,
|
||||
);
|
||||
|
||||
intakeFindMany.mockClear();
|
||||
rollupUpsert.mockClear();
|
||||
executeRaw.mockClear();
|
||||
|
||||
await recomputeMedicationComplianceForDay(
|
||||
"user-2",
|
||||
@@ -222,24 +211,53 @@ describe("recomputeMedicationComplianceForDay", () => {
|
||||
"2026-05-17",
|
||||
"America/Los_Angeles",
|
||||
);
|
||||
const laCall = intakeFindMany.mock.calls[0][0].where;
|
||||
const laInsert = bindValues(executeRaw.mock.calls[0]).filter(
|
||||
(b): b is Date => b instanceof Date,
|
||||
);
|
||||
|
||||
// Berlin 2026-05-18 starts at UTC 2026-05-17 22:00.
|
||||
expect(berlinCall.scheduledFor.gte.toISOString()).toBe(
|
||||
"2026-05-17T22:00:00.000Z",
|
||||
);
|
||||
expect(berlinInsert[0].toISOString()).toBe("2026-05-17T22:00:00.000Z");
|
||||
// LA 2026-05-17 starts at UTC 2026-05-17 07:00 (PDT is UTC-7).
|
||||
expect(laCall.scheduledFor.gte.toISOString()).toBe(
|
||||
"2026-05-17T07:00:00.000Z",
|
||||
);
|
||||
expect(laInsert[0].toISOString()).toBe("2026-05-17T07:00:00.000Z");
|
||||
});
|
||||
|
||||
it("commits the strictest aggregate under concurrent recompute calls", async () => {
|
||||
// QA F-H-02 race pin: two concurrent recomputes for the same
|
||||
// (user, medication, day) must both go through the atomic SQL
|
||||
// path. The earlier `findMany`-then-`upsert` pattern could
|
||||
// interleave A-SELECT → B-SELECT → B-UPSERT (correct) → A-UPSERT
|
||||
// (stale N-1); the atomic upsert serialises on the row lock so
|
||||
// each statement re-aggregates a snapshot taken after the prior
|
||||
// commit. The mock can't reproduce real Postgres concurrency, but
|
||||
// the test pins that BOTH callers issue the atomic statement (not
|
||||
// the legacy read-aggregate-then-upsert) so a regression to the
|
||||
// racy pattern would fail this assertion.
|
||||
await Promise.all([
|
||||
recomputeMedicationComplianceForDay(
|
||||
"user-1",
|
||||
"med-1",
|
||||
"2026-05-18",
|
||||
"Europe/Berlin",
|
||||
),
|
||||
recomputeMedicationComplianceForDay(
|
||||
"user-1",
|
||||
"med-1",
|
||||
"2026-05-18",
|
||||
"Europe/Berlin",
|
||||
),
|
||||
]);
|
||||
// 2 callers × 2 statements (INSERT + DELETE) each.
|
||||
expect(executeRaw).toHaveBeenCalledTimes(4);
|
||||
// No legacy intake-findMany was issued — the atomic path replaces it.
|
||||
expect(intakeFindMany).not.toHaveBeenCalled();
|
||||
// No legacy rollupUpsert / rollupDeleteMany either.
|
||||
expect(rollupUpsert).not.toHaveBeenCalled();
|
||||
expect(rollupDeleteMany).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("recomputeMedicationComplianceForEvent", () => {
|
||||
it("derives the dayKey from scheduledFor + tz then dispatches", async () => {
|
||||
intakeFindMany.mockResolvedValue([
|
||||
{ takenAt: new Date(), skipped: false },
|
||||
]);
|
||||
// 2026-05-17T23:30:00Z → Berlin day 2026-05-18.
|
||||
await recomputeMedicationComplianceForEvent({
|
||||
userId: "user-1",
|
||||
@@ -247,12 +265,14 @@ describe("recomputeMedicationComplianceForEvent", () => {
|
||||
scheduledFor: new Date("2026-05-17T23:30:00.000Z"),
|
||||
tz: "Europe/Berlin",
|
||||
});
|
||||
const call = rollupUpsert.mock.calls[0][0];
|
||||
expect(call.create.day).toBe("2026-05-18");
|
||||
// The atomic upsert binds the dayKey as one of the positional
|
||||
// arguments; assert "2026-05-18" reached the SQL layer.
|
||||
const insertBinds = executeRaw.mock.calls[0].slice(1);
|
||||
expect(insertBinds).toContain("2026-05-18");
|
||||
});
|
||||
|
||||
it("swallows recompute errors so the parent write never blocks", async () => {
|
||||
intakeFindMany.mockRejectedValue(new Error("DB melted"));
|
||||
executeRaw.mockRejectedValueOnce(new Error("DB melted"));
|
||||
|
||||
await expect(
|
||||
recomputeMedicationComplianceForEvent({
|
||||
@@ -373,9 +393,6 @@ describe("recomputeUserMedicationCompliance", () => {
|
||||
{ medication_id: "med-1", day: "2026-05-18" },
|
||||
{ medication_id: "med-2", day: "2026-05-18" },
|
||||
]);
|
||||
intakeFindMany.mockResolvedValue([
|
||||
{ takenAt: new Date(), skipped: false },
|
||||
]);
|
||||
|
||||
const result = await recomputeUserMedicationCompliance(
|
||||
"user-1",
|
||||
@@ -384,7 +401,10 @@ describe("recomputeUserMedicationCompliance", () => {
|
||||
);
|
||||
|
||||
expect(result.rowsUpserted).toBe(3);
|
||||
expect(rollupUpsert).toHaveBeenCalledTimes(3);
|
||||
// Each (medication, day) pair issues two atomic SQL statements:
|
||||
// the INSERT … ON CONFLICT upsert and the companion DELETE that
|
||||
// gates on `NOT EXISTS`.
|
||||
expect(executeRaw).toHaveBeenCalledTimes(6);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -198,57 +198,64 @@ export async function recomputeMedicationComplianceForDay(
|
||||
const dstSafeEnd = startOfDayUtcInTz(nextDayKey, safeTz);
|
||||
const windowEnd = dstSafeEnd.getTime() > start.getTime() ? dstSafeEnd : end;
|
||||
|
||||
const events = await client.medicationIntakeEvent.findMany({
|
||||
where: {
|
||||
userId,
|
||||
medicationId,
|
||||
scheduledFor: { gte: start, lt: windowEnd },
|
||||
},
|
||||
select: { takenAt: true, skipped: true },
|
||||
});
|
||||
|
||||
let scheduled = 0;
|
||||
let taken = 0;
|
||||
let skipped = 0;
|
||||
for (const evt of events) {
|
||||
scheduled += 1;
|
||||
if (evt.skipped) {
|
||||
skipped += 1;
|
||||
} else if (evt.takenAt) {
|
||||
taken += 1;
|
||||
}
|
||||
}
|
||||
|
||||
if (scheduled === 0) {
|
||||
// No events landed inside the user-day window after the mutation —
|
||||
// the row would otherwise stay around with a stale (1, 1, 0) tuple
|
||||
// from a now-deleted intake event. Delete the rollup row so the
|
||||
// reader returns the trailing-window zero-default for this day.
|
||||
await client.medicationComplianceRollup.deleteMany({
|
||||
where: { userId, medicationId, day: dayKey },
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
await client.medicationComplianceRollup.upsert({
|
||||
where: {
|
||||
userId_medicationId_day: { userId, medicationId, day: dayKey },
|
||||
},
|
||||
update: {
|
||||
scheduled,
|
||||
taken,
|
||||
skipped,
|
||||
computedAt: new Date(),
|
||||
},
|
||||
create: {
|
||||
userId,
|
||||
medicationId,
|
||||
day: dayKey,
|
||||
scheduled,
|
||||
taken,
|
||||
skipped,
|
||||
},
|
||||
});
|
||||
// v1.4.39 hotfix (QA F-H-02): atomic upsert closes the race window
|
||||
// between the prior `SELECT … aggregate then UPSERT` pattern. Two
|
||||
// concurrent writes for the same (user, medication, day) used to
|
||||
// interleave A-SELECT → B-SELECT → B-UPSERT (correct) → A-UPSERT
|
||||
// (stale N-1). The INSERT … SELECT re-aggregates inside the upsert
|
||||
// statement so each write commits with a snapshot taken after the
|
||||
// prior commit released its row lock; the trailing DELETE handles
|
||||
// the "all events for the day were removed" case in the same shot.
|
||||
//
|
||||
// Both statements key on (user_id, medication_id, day) so they
|
||||
// serialise behind the row lock the rollup table's primary key
|
||||
// provides. The DELETE runs second so it cannot clobber a row the
|
||||
// INSERT just wrote — when the day still has events the
|
||||
// `WHERE NOT EXISTS` predicate matches zero rows.
|
||||
await client.$executeRaw`
|
||||
WITH aggregate AS (
|
||||
SELECT
|
||||
COUNT(*)::int AS scheduled,
|
||||
SUM(CASE WHEN "skipped" THEN 0 WHEN "taken_at" IS NOT NULL THEN 1 ELSE 0 END)::int AS taken,
|
||||
SUM(CASE WHEN "skipped" THEN 1 ELSE 0 END)::int AS skipped
|
||||
FROM "medication_intake_events"
|
||||
WHERE "user_id" = ${userId}
|
||||
AND "medication_id" = ${medicationId}
|
||||
AND "scheduled_for" >= ${start}
|
||||
AND "scheduled_for" < ${windowEnd}
|
||||
)
|
||||
INSERT INTO "medication_compliance_rollups"
|
||||
("user_id", "medication_id", "day", "scheduled", "taken", "skipped", "computed_at")
|
||||
SELECT
|
||||
${userId},
|
||||
${medicationId},
|
||||
${dayKey},
|
||||
aggregate.scheduled,
|
||||
aggregate.taken,
|
||||
aggregate.skipped,
|
||||
NOW()
|
||||
FROM aggregate
|
||||
WHERE aggregate.scheduled > 0
|
||||
ON CONFLICT ("user_id", "medication_id", "day") DO UPDATE
|
||||
SET "scheduled" = EXCLUDED."scheduled",
|
||||
"taken" = EXCLUDED."taken",
|
||||
"skipped" = EXCLUDED."skipped",
|
||||
"computed_at" = EXCLUDED."computed_at"
|
||||
`;
|
||||
await client.$executeRaw`
|
||||
DELETE FROM "medication_compliance_rollups"
|
||||
WHERE "user_id" = ${userId}
|
||||
AND "medication_id" = ${medicationId}
|
||||
AND "day" = ${dayKey}
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM "medication_intake_events"
|
||||
WHERE "user_id" = ${userId}
|
||||
AND "medication_id" = ${medicationId}
|
||||
AND "scheduled_for" >= ${start}
|
||||
AND "scheduled_for" < ${windowEnd}
|
||||
)
|
||||
`;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -21,6 +21,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
const mocks = vi.hoisted(() => ({
|
||||
queryRaw: vi.fn(),
|
||||
queryRawUnsafe: vi.fn(),
|
||||
executeRaw: vi.fn(),
|
||||
upsert: vi.fn(),
|
||||
deleteMany: vi.fn(),
|
||||
findFirst: vi.fn(),
|
||||
@@ -34,6 +35,7 @@ vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
$queryRaw: mocks.queryRaw,
|
||||
$queryRawUnsafe: mocks.queryRawUnsafe,
|
||||
$executeRaw: mocks.executeRaw,
|
||||
moodEntryRollup: {
|
||||
upsert: mocks.upsert,
|
||||
deleteMany: mocks.deleteMany,
|
||||
@@ -59,6 +61,7 @@ import { annotate } from "@/lib/logging/context";
|
||||
const {
|
||||
queryRaw,
|
||||
queryRawUnsafe,
|
||||
executeRaw,
|
||||
upsert,
|
||||
deleteMany,
|
||||
findFirst,
|
||||
@@ -85,6 +88,8 @@ import {
|
||||
beforeEach(() => {
|
||||
queryRaw.mockReset();
|
||||
queryRawUnsafe.mockReset();
|
||||
executeRaw.mockReset();
|
||||
executeRaw.mockResolvedValue(0);
|
||||
upsert.mockReset();
|
||||
deleteMany.mockReset();
|
||||
findFirst.mockReset();
|
||||
@@ -102,17 +107,15 @@ afterEach(() => {
|
||||
});
|
||||
|
||||
describe("recomputeMoodBucketsForEntry", () => {
|
||||
it("upserts the DAY rollup synchronously and enqueues WEEK/MONTH/YEAR", async () => {
|
||||
queryRawUnsafe.mockResolvedValueOnce([
|
||||
{
|
||||
bucket_start: new Date("2026-05-10T00:00:00.000Z"),
|
||||
count: BigInt(3),
|
||||
mean: 4.0,
|
||||
min_score: 3,
|
||||
max_score: 5,
|
||||
sd: 0.82,
|
||||
},
|
||||
]);
|
||||
function bindValues(call: unknown[]): unknown[] {
|
||||
return call.slice(1);
|
||||
}
|
||||
|
||||
it("runs the atomic DAY upsert + DELETE pair and fires WEEK/MONTH/YEAR enqueues", async () => {
|
||||
// QA F-H-02 (v1.4.39): the DAY pass is now a single atomic SQL
|
||||
// statement (INSERT … SELECT … ON CONFLICT) plus a paired DELETE
|
||||
// that gates on NOT EXISTS. No JS-side aggregate, no per-row
|
||||
// upsert through the Prisma client.
|
||||
getGlobalBossMock.mockReturnValue({ send: bossSend });
|
||||
bossSend.mockResolvedValue("job-id");
|
||||
|
||||
@@ -121,22 +124,23 @@ describe("recomputeMoodBucketsForEntry", () => {
|
||||
new Date("2026-05-10T14:30:00.000Z"),
|
||||
);
|
||||
|
||||
// DAY pass — single upsert against the rollup table.
|
||||
expect(upsert).toHaveBeenCalledTimes(1);
|
||||
const upsertArg = upsert.mock.calls[0][0];
|
||||
expect(upsertArg.where.userId_granularity_bucketStart.userId).toBe(
|
||||
"user-1",
|
||||
);
|
||||
expect(upsertArg.where.userId_granularity_bucketStart.granularity).toBe(
|
||||
"DAY",
|
||||
);
|
||||
expect(upsertArg.create.count).toBe(3);
|
||||
expect(upsertArg.create.mean).toBe(4);
|
||||
expect(upsertArg.create.minScore).toBe(3);
|
||||
expect(upsertArg.create.maxScore).toBe(5);
|
||||
expect(upsertArg.create.sd).toBeCloseTo(0.82);
|
||||
expect(executeRaw).toHaveBeenCalledTimes(2);
|
||||
const insertBinds = bindValues(executeRaw.mock.calls[0]);
|
||||
expect(insertBinds).toContain("user-1");
|
||||
// bucket_start is bound twice (INSERT row + INSERT SELECT predicate
|
||||
// are both inside the same CTE shape via the same parameter).
|
||||
const dayStart = new Date("2026-05-10T00:00:00.000Z");
|
||||
expect(
|
||||
(insertBinds.filter((b) => b instanceof Date) as Date[]).some(
|
||||
(d) => d.getTime() === dayStart.getTime(),
|
||||
),
|
||||
).toBe(true);
|
||||
|
||||
// QA F-H-03 — the enqueue is fire-and-forget; the response path
|
||||
// never awaits the boss.send call. Wait a tick for the
|
||||
// floating promise to flush before asserting.
|
||||
await new Promise((r) => setImmediate(r));
|
||||
|
||||
// WEEK / MONTH / YEAR — three enqueues against the worker queue.
|
||||
expect(bossSend).toHaveBeenCalledTimes(3);
|
||||
for (const call of bossSend.mock.calls) {
|
||||
expect(call[0]).toBe(MOOD_ROLLUP_RECOMPUTE_QUEUE);
|
||||
@@ -145,9 +149,7 @@ describe("recomputeMoodBucketsForEntry", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("deletes the DAY row when the post-mutation aggregate is empty", async () => {
|
||||
// Post-delete recompute: the day now has zero mood rows.
|
||||
queryRawUnsafe.mockResolvedValueOnce([]);
|
||||
it("issues the paired DELETE so a fully-cleared day removes its row", async () => {
|
||||
getGlobalBossMock.mockReturnValue(null);
|
||||
|
||||
await recomputeMoodBucketsForEntry(
|
||||
@@ -155,30 +157,16 @@ describe("recomputeMoodBucketsForEntry", () => {
|
||||
new Date("2026-05-10T14:30:00.000Z"),
|
||||
);
|
||||
|
||||
expect(deleteMany).toHaveBeenCalledTimes(1);
|
||||
expect(deleteMany.mock.calls[0][0].where.userId).toBe("user-1");
|
||||
expect(deleteMany.mock.calls[0][0].where.granularity).toBe("DAY");
|
||||
// The DELETE always fires; its NOT EXISTS predicate matches zero
|
||||
// rows on the populated-day branch and matches the row on the
|
||||
// emptied-day branch. We can only assert the second statement
|
||||
// fired here.
|
||||
expect(executeRaw).toHaveBeenCalledTimes(2);
|
||||
expect(deleteMany).not.toHaveBeenCalled();
|
||||
expect(upsert).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("is idempotent under re-run for the same (user, day)", async () => {
|
||||
// Two consecutive calls with the same aggregate output should
|
||||
// result in two upserts that both carry the same payload — the
|
||||
// composite PK absorbs the second write into a no-op refresh on
|
||||
// the database side, but the populator must keep firing the
|
||||
// upsert each time (no in-memory dedup) so a real Postgres-side
|
||||
// change is picked up on the next call.
|
||||
const aggregate = [
|
||||
{
|
||||
bucket_start: new Date("2026-05-10T00:00:00.000Z"),
|
||||
count: BigInt(2),
|
||||
mean: 4.5,
|
||||
min_score: 4,
|
||||
max_score: 5,
|
||||
sd: 0.5,
|
||||
},
|
||||
];
|
||||
queryRawUnsafe.mockResolvedValue(aggregate);
|
||||
getGlobalBossMock.mockReturnValue(null);
|
||||
|
||||
await recomputeMoodBucketsForEntry(
|
||||
@@ -190,13 +178,64 @@ describe("recomputeMoodBucketsForEntry", () => {
|
||||
new Date("2026-05-10T14:30:00.000Z"),
|
||||
);
|
||||
|
||||
expect(upsert).toHaveBeenCalledTimes(2);
|
||||
const firstPayload = upsert.mock.calls[0][0].create;
|
||||
const secondPayload = upsert.mock.calls[1][0].create;
|
||||
expect(firstPayload.count).toBe(secondPayload.count);
|
||||
expect(firstPayload.mean).toBe(secondPayload.mean);
|
||||
expect(firstPayload.minScore).toBe(secondPayload.minScore);
|
||||
expect(firstPayload.maxScore).toBe(secondPayload.maxScore);
|
||||
expect(executeRaw).toHaveBeenCalledTimes(4);
|
||||
// Bind values for the INSERT statement match across both runs.
|
||||
expect(bindValues(executeRaw.mock.calls[0])).toEqual(
|
||||
bindValues(executeRaw.mock.calls[2]),
|
||||
);
|
||||
// Bind values for the DELETE statement match across both runs.
|
||||
expect(bindValues(executeRaw.mock.calls[1])).toEqual(
|
||||
bindValues(executeRaw.mock.calls[3]),
|
||||
);
|
||||
});
|
||||
|
||||
it("commits the strictest aggregate under concurrent recompute calls", async () => {
|
||||
// QA F-H-02 race pin: two concurrent recomputes for the same
|
||||
// (user, day) must both go through the atomic SQL path so each
|
||||
// statement re-aggregates a snapshot taken after the prior commit
|
||||
// released its row lock.
|
||||
getGlobalBossMock.mockReturnValue(null);
|
||||
|
||||
await Promise.all([
|
||||
recomputeMoodBucketsForEntry(
|
||||
"user-1",
|
||||
new Date("2026-05-10T14:30:00.000Z"),
|
||||
),
|
||||
recomputeMoodBucketsForEntry(
|
||||
"user-1",
|
||||
new Date("2026-05-10T14:30:00.000Z"),
|
||||
),
|
||||
]);
|
||||
// 2 callers × 2 statements (INSERT + DELETE) each.
|
||||
expect(executeRaw).toHaveBeenCalledTimes(4);
|
||||
// No legacy JS-side aggregate ($queryRawUnsafe) fired on the DAY
|
||||
// pass; no Prisma-API upsert / deleteMany were issued either.
|
||||
expect(queryRawUnsafe).not.toHaveBeenCalled();
|
||||
expect(upsert).not.toHaveBeenCalled();
|
||||
expect(deleteMany).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not await the WEEK/MONTH/YEAR enqueue on the write-path response", async () => {
|
||||
// QA F-H-03 (v1.4.39): the outer Promise.all is fire-and-forget so
|
||||
// a slow pg-boss send never serialises the user's mood-write
|
||||
// response. Pin that the helper resolves before the enqueue
|
||||
// promise settles.
|
||||
let resolveEnqueue: (() => void) | null = null;
|
||||
const enqueueGate = new Promise<string>((resolve) => {
|
||||
resolveEnqueue = () => resolve("job-late");
|
||||
});
|
||||
getGlobalBossMock.mockReturnValue({ send: () => enqueueGate });
|
||||
|
||||
await recomputeMoodBucketsForEntry(
|
||||
"user-1",
|
||||
new Date("2026-05-10T14:30:00.000Z"),
|
||||
);
|
||||
|
||||
// The helper has resolved even though the boss.send promise is
|
||||
// still pending. Resolve it after the assertion so the test does
|
||||
// not leak a pending promise across the suite.
|
||||
resolveEnqueue!();
|
||||
await enqueueGate;
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
+76
-27
@@ -115,7 +115,7 @@ interface MoodRollupRow {
|
||||
*/
|
||||
type PrismaTxOrClient = Pick<
|
||||
PrismaClient,
|
||||
"$queryRawUnsafe" | "moodEntryRollup"
|
||||
"$queryRawUnsafe" | "$executeRaw" | "moodEntryRollup"
|
||||
>;
|
||||
|
||||
/**
|
||||
@@ -176,37 +176,86 @@ export async function recomputeMoodBucketsForEntry(
|
||||
tx?: PrismaTxOrClient,
|
||||
): Promise<void> {
|
||||
const client = tx ?? prisma;
|
||||
// DAY pass — synchronous. Aggregate over the affected day window.
|
||||
// DAY pass — synchronous. v1.4.39 hotfix (QA F-H-02): one atomic SQL
|
||||
// statement re-aggregates inside the upsert subquery, closing the
|
||||
// race window where two concurrent writes for the same (user, day)
|
||||
// could interleave A-SELECT → B-SELECT → B-UPSERT (correct) →
|
||||
// A-UPSERT (stale N-1). The trailing DELETE handles the "all entries
|
||||
// for the day were removed" case in the same shot.
|
||||
const dayStart = startOfUtcDay(moodLoggedAt);
|
||||
const dayEnd = new Date(dayStart.getTime() + 24 * 60 * 60 * 1000);
|
||||
const rows = await runMoodRollupAggregate(client, {
|
||||
userId,
|
||||
granularity: "DAY",
|
||||
from: dayStart,
|
||||
to: dayEnd,
|
||||
});
|
||||
if (rows.length === 0) {
|
||||
// No mood rows in this day after the mutation (e.g. the DELETE
|
||||
// path removed the last entry) — drop the rollup row if it exists.
|
||||
await client.moodEntryRollup.deleteMany({
|
||||
where: {
|
||||
userId,
|
||||
granularity: "DAY",
|
||||
bucketStart: dayStart,
|
||||
},
|
||||
});
|
||||
} else {
|
||||
await persistMoodRollupRows(client, userId, "DAY", rows);
|
||||
}
|
||||
await client.$executeRaw`
|
||||
WITH aggregate AS (
|
||||
SELECT
|
||||
COUNT(*)::int AS count,
|
||||
AVG(m."score")::double precision AS mean,
|
||||
MIN(m."score")::integer AS min_score,
|
||||
MAX(m."score")::integer AS max_score,
|
||||
STDDEV_POP(m."score")::double precision AS sd
|
||||
FROM "mood_entries" m
|
||||
WHERE m."user_id" = ${userId}
|
||||
AND m."mood_logged_at" >= ${dayStart}
|
||||
AND m."mood_logged_at" < ${dayEnd}
|
||||
)
|
||||
INSERT INTO "mood_entry_rollups"
|
||||
("user_id", "granularity", "bucket_start", "count", "mean", "min_score", "max_score", "sd", "computed_at")
|
||||
SELECT
|
||||
${userId},
|
||||
'DAY',
|
||||
${dayStart},
|
||||
aggregate.count,
|
||||
aggregate.mean,
|
||||
aggregate.min_score,
|
||||
aggregate.max_score,
|
||||
aggregate.sd,
|
||||
NOW()
|
||||
FROM aggregate
|
||||
WHERE aggregate.count > 0
|
||||
ON CONFLICT ("user_id", "granularity", "bucket_start") DO UPDATE
|
||||
SET "count" = EXCLUDED."count",
|
||||
"mean" = EXCLUDED."mean",
|
||||
"min_score" = EXCLUDED."min_score",
|
||||
"max_score" = EXCLUDED."max_score",
|
||||
"sd" = EXCLUDED."sd",
|
||||
"computed_at" = EXCLUDED."computed_at"
|
||||
`;
|
||||
await client.$executeRaw`
|
||||
DELETE FROM "mood_entry_rollups"
|
||||
WHERE "user_id" = ${userId}
|
||||
AND "granularity" = 'DAY'
|
||||
AND "bucket_start" = ${dayStart}
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM "mood_entries" m
|
||||
WHERE m."user_id" = ${userId}
|
||||
AND m."mood_logged_at" >= ${dayStart}
|
||||
AND m."mood_logged_at" < ${dayEnd}
|
||||
)
|
||||
`;
|
||||
|
||||
// WEEK / MONTH / YEAR — enqueue. Each granularity covers the bucket
|
||||
// that contains `moodLoggedAt` so the worker only re-aggregates the
|
||||
// bucket the write could have changed. Fan-out in parallel — the
|
||||
// singleton key per granularity prevents duplicate jobs.
|
||||
await Promise.all(
|
||||
// WEEK / MONTH / YEAR — fire-and-forget enqueue. QA F-H-03 (v1.4.39):
|
||||
// the outer await on this Promise.all used to serialise the write-path
|
||||
// response on pg-boss latency. The DAY pass the read path depends on
|
||||
// has already committed synchronously above; the queue is a hint, not
|
||||
// a write-path invariant, so dropping the await keeps the response
|
||||
// off the pg-boss critical path. Errors stay caught inside
|
||||
// `enqueueMoodRollupRecompute` so the unhandled-rejection logger
|
||||
// captures any pg-boss failure.
|
||||
void Promise.all(
|
||||
ASYNC_GRANULARITIES.map((granularity) => {
|
||||
const { from, to } = bucketSpan(moodLoggedAt, granularity);
|
||||
return enqueueMoodRollupRecompute({ userId, granularity, from, to });
|
||||
return enqueueMoodRollupRecompute({ userId, granularity, from, to }).catch(
|
||||
(err: unknown) => {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
annotate({
|
||||
meta: {
|
||||
mood_rollup_enqueue_failed: true,
|
||||
mood_rollup_enqueue_error: message,
|
||||
},
|
||||
});
|
||||
console.error("[mood-rollups] async enqueue failed:", message);
|
||||
},
|
||||
);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user