From 362f87d4c4d6cf0e11cf0b304637dbe2a498dd15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 18 May 2026 12:51:31 +0200 Subject: [PATCH] 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. --- .../__tests__/compliance-rollups.test.ts | 180 ++++++++++-------- src/lib/medications/compliance-rollups.ts | 109 ++++++----- src/lib/mood/__tests__/rollups.test.ts | 151 +++++++++------ src/lib/mood/rollups.ts | 103 +++++++--- 4 files changed, 329 insertions(+), 214 deletions(-) diff --git a/src/lib/medications/__tests__/compliance-rollups.test.ts b/src/lib/medications/__tests__/compliance-rollups.test.ts index 7b6374b03..effee8ef7 100644 --- a/src/lib/medications/__tests__/compliance-rollups.test.ts +++ b/src/lib/medications/__tests__/compliance-rollups.test.ts @@ -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); }); }); diff --git a/src/lib/medications/compliance-rollups.ts b/src/lib/medications/compliance-rollups.ts index 2b25c96a3..27b5ab4c4 100644 --- a/src/lib/medications/compliance-rollups.ts +++ b/src/lib/medications/compliance-rollups.ts @@ -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} + ) + `; } /** diff --git a/src/lib/mood/__tests__/rollups.test.ts b/src/lib/mood/__tests__/rollups.test.ts index 8239eb9a2..5b5d6ff99 100644 --- a/src/lib/mood/__tests__/rollups.test.ts +++ b/src/lib/mood/__tests__/rollups.test.ts @@ -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((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; }); }); diff --git a/src/lib/mood/rollups.ts b/src/lib/mood/rollups.ts index b3f2000fe..4601db8e8 100644 --- a/src/lib/mood/rollups.ts +++ b/src/lib/mood/rollups.ts @@ -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 { 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); + }, + ); }), ); }