diff --git a/src/lib/rollups/__tests__/mood-rollups.test.ts b/src/lib/rollups/__tests__/mood-rollups.test.ts index 6f6aa2ec7..53f8196de 100644 --- a/src/lib/rollups/__tests__/mood-rollups.test.ts +++ b/src/lib/rollups/__tests__/mood-rollups.test.ts @@ -387,33 +387,39 @@ describe("recomputeUserMoodRollups replace (delete-then-refold, v1.32.12)", () = sd: 0.5, }; queryRawUnsafe.mockResolvedValue([aggRow]); - upsert.mockResolvedValue({}); + // The fold writes every bucket in one multi-row upsert; the values of + // that statement are the bucket key and the folded stats, in column + // order, with the run's computedAt last. + const writtenRows = () => + executeRaw.mock.calls.map((call) => + ((call[1] as { values: unknown[] }).values ?? []).slice(0, -1), + ); await recomputeUserMoodRollups("user-1", { granularities: ["DAY"], replace: true, }); - const first = upsert.mock.calls.map((c) => c[0] as Record); - upsert.mockClear(); + const first = writtenRows(); + executeRaw.mockClear(); await recomputeUserMoodRollups("user-1", { granularities: ["DAY"], replace: true, }); - const second = upsert.mock.calls.map( - (c) => c[0] as Record, - ); + const second = writtenRows(); expect(first).toHaveLength(1); - expect(second).toHaveLength(1); - // Same bucket key + same folded stats across both rebuilds (only the - // computedAt wall-clock differs, which we don't compare). - expect(first[0].where).toEqual(second[0].where); - expect((first[0].create as { count: number }).count).toBe( - (second[0].create as { count: number }).count, - ); - expect((first[0].create as { mean: number }).mean).toBe( - (second[0].create as { mean: number }).mean, - ); + expect(second).toEqual(first); + expect(first[0]).toEqual([ + "user-1", + "DAY", + new Date("2026-05-10T00:00:00.000Z"), + 3, + 4.2, + 3, + 5, + 0.5, + ]); + expect(upsert).not.toHaveBeenCalled(); }); }); diff --git a/src/lib/rollups/mood-rollups.ts b/src/lib/rollups/mood-rollups.ts index b7fccea27..4dfe341e7 100644 --- a/src/lib/rollups/mood-rollups.ts +++ b/src/lib/rollups/mood-rollups.ts @@ -51,10 +51,10 @@ import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { annotate } from "@/lib/logging/context"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; -import type { +import { Prisma, - PrismaClient, - RollupGranularity, + type PrismaClient, + type RollupGranularity, } from "@/generated/prisma/client"; import { dayKeyAsUtcMidnight } from "@/lib/tz/date-only"; @@ -668,42 +668,32 @@ async function persistMoodRollupRows( rows: MoodRollupRow[], ): Promise { if (rows.length === 0) return 0; + // One multi-row upsert per chunk. An upsert per bucket cost a statement for + // every day of the account's history on the read that refreshed the tier + // (the dashboard summary among them), and the count moved with the + // calendar. A single statement is also atomic on its own, so it is safe + // inside a caller's interactive transaction or outside one. const CHUNK = 500; + const computedAt = new Date(); let touched = 0; for (let i = 0; i < rows.length; i += CHUNK) { const slice = rows.slice(i, i + CHUNK); - // Run each upsert sequentially through the supplied client (which - // may be a transactional handle). Wrapping in `$transaction` here - // would crash when the caller already holds an interactive - // transaction, so we rely on the parent caller (or atomicity of - // the single upsert) for crash safety. The hot path is a single - // row per DAY recompute, so the round-trip cost is negligible. - for (const row of slice) { - const payload = { - count: Number(row.count), - mean: row.mean ?? 0, - minScore: row.min_score ?? 0, - maxScore: row.max_score ?? 0, - sd: row.sd, - computedAt: new Date(), - } satisfies Prisma.MoodEntryRollupUpdateInput; - await client.moodEntryRollup.upsert({ - where: { - userId_granularity_bucketStart: { - userId, - granularity, - bucketStart: row.bucket_start, - }, - }, - create: { - userId, - granularity, - bucketStart: row.bucket_start, - ...payload, - }, - update: payload, - }); - } + const values = slice.map( + (row) => + Prisma.sql`(${userId}, ${granularity}::"measurement_rollup_granularity", ${row.bucket_start}, ${Number(row.count)}, ${row.mean ?? 0}, ${row.min_score ?? 0}, ${row.max_score ?? 0}, ${row.sd}, ${computedAt})`, + ); + await client.$executeRaw` + INSERT INTO "mood_entry_rollups" + ("user_id", "granularity", "bucket_start", "count", "mean", "min_score", "max_score", "sd", "computed_at") + VALUES ${Prisma.join(values)} + 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" + `; touched += slice.length; } return touched; diff --git a/tests/integration/query-budget-hot-routes.test.ts b/tests/integration/query-budget-hot-routes.test.ts index 3bab1970b..cb0390f26 100644 --- a/tests/integration/query-budget-hot-routes.test.ts +++ b/tests/integration/query-budget-hot-routes.test.ts @@ -67,7 +67,7 @@ const DAY_MS = 86_400_000; */ const BUDGET = { authMe: 13, - dashboardSummary: 84, + dashboardSummary: 28, dashboardSnapshot: 99, series30: 6, series365: 5,