From 6742b4f90bed1a5bd98ba0dc715548b06661c9f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 18 May 2026 12:16:30 +0200 Subject: [PATCH] feat(mood-rollups): per-write hook and reader helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirrors the measurement-rollups tier for the mood domain. Adds the write-path hook (synchronous DAY upsert + async WEEK/MONTH/YEAR enqueue), the bounded DAY reader, the warm-up helper, and the boot-time backfill discovery. Closes the §2.3 finding in the v1.4.38 perf analysis where /api/mood/analytics walked every MoodEntry on every cold mount. --- src/lib/mood/rollups.ts | 621 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 621 insertions(+) create mode 100644 src/lib/mood/rollups.ts diff --git a/src/lib/mood/rollups.ts b/src/lib/mood/rollups.ts new file mode 100644 index 000000000..b3f2000fe --- /dev/null +++ b/src/lib/mood/rollups.ts @@ -0,0 +1,621 @@ +/** + * v1.4.39 W-MOOD — persistent mood-rollup populator. + * + * Sits behind the 60-s `caches.moodAnalytics` LRU as the second cache + * tier on `/api/mood/analytics`. One row per + * `(userId, granularity, bucketStart)` in `mood_entry_rollups`, + * carrying `count` / `mean` / `min_score` / `max_score` / `sd` + * (population) folded via Postgres `AVG` / `MIN` / `MAX` / `STDDEV_POP`. + * + * The audit at `.planning/round-v1438-perf-analysis.md` §2.3 traced + * the 12.7 s cold mount on `/api/mood/analytics` to the unbounded + * `prisma.moodEntry.findMany({ where: { userId } })` that walked + * every mood row the user has ever written into Node only to + * bucket-average them in JS. This module mirrors the well-trodden + * `src/lib/measurements/rollups.ts` pattern so the read path can swap + * to a bounded `prisma.moodEntryRollup.findMany` with the daily mean + * already materialised. + * + * Write hooks: + * - Every mood-entry create/update/delete fires + * `recomputeMoodBucketsForEntry`. The DAY pass runs inline + * (cheap — single `(user, day)` aggregate) so the next read sees + * the fresh row without waiting on pg-boss. + * - WEEK / MONTH / YEAR are enqueued onto `mood-rollup-recompute` + * for the worker to fold. No current reader consumes them; the + * buckets exist so future cross-granularity readers ship without + * a backfill step. + * - Bulk endpoints (`/api/mood-entries/bulk`, restore) call + * `recomputeUserMoodRollups(userId)` once at the end of the + * batch instead of firing N per-row hooks. + * + * Reads: + * - `readMoodDayRollups(userId, since)` returns the persisted DAY + * rows. The reader-side fallback in `/api/mood/analytics` checks + * the rollup count and falls through to the legacy live findMany + * once when the user has mood entries but zero rollup rows; + * `enqueueBootTimeMoodRollupBackfill` mints the rollups on the + * next worker boot so subsequent reads converge on the fast path. + */ +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { annotate } from "@/lib/logging/context"; +import type { + Prisma, + PrismaClient, + RollupGranularity, +} from "@/generated/prisma/client"; + +/** Granularities that recompute asynchronously via pg-boss. */ +const ASYNC_GRANULARITIES: RollupGranularity[] = ["WEEK", "MONTH", "YEAR"]; + +/** All granularities the rollup table tracks. */ +export const ALL_MOOD_GRANULARITIES: RollupGranularity[] = [ + "DAY", + "WEEK", + "MONTH", + "YEAR", +]; + +/** pg-boss queue name for per-bucket recompute jobs. */ +export const MOOD_ROLLUP_RECOMPUTE_QUEUE = "mood-rollup-recompute"; + +/** Worker concurrency cap — recomputes are read-heavy on Postgres. */ +export const MOOD_ROLLUP_RECOMPUTE_CONCURRENCY = 2; + +/** pg-boss queue name for the boot-time backfill discovery. */ +export const MOOD_ROLLUP_FULL_BACKFILL_QUEUE = "mood-rollup-full-backfill"; + +/** Boot-fold worker concurrency. Serial so we don't crowd the pool. */ +export const MOOD_ROLLUP_FULL_BACKFILL_CONCURRENCY = 1; + +/** Payload the worker queue carries. */ +export interface MoodRollupRecomputePayload { + userId: string; + granularity: RollupGranularity; + /** ISO-8601 — the worker parses through `new Date(value)`. */ + from: string; + /** ISO-8601 — the worker parses through `new Date(value)`. */ + to: string; + /** Wall-clock kick-off for debugging only. */ + enqueuedAt: string; +} + +/** Payload the boot-time backfill helper enqueues. */ +export interface MoodRollupFullBackfillPayload { + userId: string; + /** Wall-clock kick-off for debugging only. */ + enqueuedAt: string; +} + +/** Postgres `date_trunc` unit per granularity. */ +const DATE_TRUNC_UNIT: Record = { + DAY: "day", + WEEK: "week", + MONTH: "month", + YEAR: "year", +}; + +/** Raw row shape returned by the aggregate `$queryRawUnsafe`. */ +interface MoodRollupRow { + bucket_start: Date; + count: bigint; + mean: number | null; + min_score: number | null; + max_score: number | null; + sd: number | null; +} + +/** + * Tx-or-client surface the writer accepts. Mirrors the + * `measurement-rollups` write-path contract: when called from inside a + * `prisma.$transaction(async tx => …)` block, callers pass the `tx` + * client so the rollup write commits atomically with the parent + * mutation. + */ +type PrismaTxOrClient = Pick< + PrismaClient, + "$queryRawUnsafe" | "moodEntryRollup" +>; + +/** + * Recompute the rollup buckets for one user. Bounded by the optional + * `granularities` / `from` / `to` filters; defaults cover every + * granularity and the trailing 5 years. + * + * Idempotent: rows are upserted under the + * `(userId, granularity, bucketStart)` composite primary key, so + * re-running the call is a no-op for unchanged buckets and a refresh + * for changed ones. + */ +export async function recomputeUserMoodRollups( + userId: string, + opts: { + granularities?: RollupGranularity[]; + from?: Date; + to?: Date; + } = {}, +): Promise<{ rowsUpserted: number; durationMs: number }> { + const startedAt = Date.now(); + const granularities = opts.granularities ?? ALL_MOOD_GRANULARITIES; + const fiveYearsAgo = new Date(Date.now() - 5 * 365 * 24 * 60 * 60 * 1000); + const from = opts.from ?? fiveYearsAgo; + const to = opts.to ?? new Date(); + + let rowsUpserted = 0; + + for (const granularity of granularities) { + const rows = await runMoodRollupAggregate(prisma, { + userId, + granularity, + from, + to, + }); + if (rows.length === 0) continue; + + rowsUpserted += await persistMoodRollupRows(prisma, userId, granularity, rows); + } + + return { rowsUpserted, durationMs: Date.now() - startedAt }; +} + +/** + * Synchronous DAY-bucket recompute for the write-path hooks plus an + * async enqueue of the WEEK / MONTH / YEAR recomputes. + * + * Scoped to the affected `(userId, day)` tuple — cheap, and avoids + * invalidating buckets that haven't changed. + * + * Pass `tx` when called from inside a `prisma.$transaction` block so + * the rollup write commits atomically with the parent mutation. + * Otherwise the global `prisma` client is used. + */ +export async function recomputeMoodBucketsForEntry( + userId: string, + moodLoggedAt: Date, + tx?: PrismaTxOrClient, +): Promise { + const client = tx ?? prisma; + // DAY pass — synchronous. Aggregate over the affected day window. + 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); + } + + // 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( + ASYNC_GRANULARITIES.map((granularity) => { + const { from, to } = bucketSpan(moodLoggedAt, granularity); + return enqueueMoodRollupRecompute({ userId, granularity, from, to }); + }), + ); +} + +/** + * Best-effort pg-boss enqueue. Silent no-op when no boss is attached + * (route running in a context without the worker — typically the test + * suite). The reader-side fall-through still produces fresh data on a + * cache miss, so missing the enqueue is at-most a one-read cost + * rather than a correctness bug. + */ +export async function enqueueMoodRollupRecompute(input: { + userId: string; + granularity: RollupGranularity; + from: Date; + to: Date; +}): Promise { + const boss = getGlobalBoss(); + if (!boss) return; + const payload: MoodRollupRecomputePayload = { + userId: input.userId, + granularity: input.granularity, + from: input.from.toISOString(), + to: input.to.toISOString(), + enqueuedAt: new Date().toISOString(), + }; + await boss.send(MOOD_ROLLUP_RECOMPUTE_QUEUE, payload, { + retryLimit: 3, + retryDelay: 30, + retryBackoff: true, + // Coalesce duplicate enqueues for the same bucket key — two writes + // inside the same week share one queued job. + singletonKey: `${input.userId}|${input.granularity}|${input.from.toISOString()}`, + }); +} + +/** + * Read DAY-granularity rollup rows for one user, sorted ascending by + * `bucketStart`. The read-path swap in `/api/mood/analytics` consumes + * these directly without any extra Node-side aggregation. + * + * `since` is an inclusive lower bound. Pass a 5-year-old `Date` from + * the route to mirror the legacy "unbounded" semantics; future + * consumers can narrow the window cheaply because the rollup table + * carries one row per day rather than one per entry. + */ +export async function readMoodDayRollups( + userId: string, + since: Date, +): Promise< + Array<{ + bucketStart: Date; + count: number; + mean: number; + minScore: number; + maxScore: number; + sd: number | null; + computedAt: Date; + }> +> { + const rows = await prisma.moodEntryRollup.findMany({ + where: { + userId, + granularity: "DAY", + bucketStart: { gte: since }, + }, + orderBy: { bucketStart: "asc" }, + }); + return rows.map((r) => ({ + bucketStart: r.bucketStart, + count: r.count, + mean: r.mean, + minScore: r.minScore, + maxScore: r.maxScore, + sd: r.sd, + computedAt: r.computedAt, + })); +} + +/** + * v1.4.39 — fire-and-forget warm-up for the DAY tier. + * + * Contract: + * - If at least one DAY rollup row exists for the user and it is + * fresh (newest rollup `computed_at` ≥ newest mood entry + * `updated_at`), this is a no-op. + * - Otherwise the helper kicks off a full + * `recomputeUserMoodRollups` over the trailing 5-year window so + * the next read of `/api/mood/analytics` lands on the rollup + * tier. Mood entries are typically 1/day so the recompute is + * bounded to a few thousand rows even for power users — cheaper + * than the 12.7 s legacy live walk by orders of magnitude. + * - Best-effort: thrown errors are swallowed so the read path + * never fails because of a populator hiccup. + * - Concurrent callers for the same `userId` share one in-flight + * promise so a cold fan-out can never queue N parallel + * recompute runs. + */ +const ensureUserMoodRollupsFreshInFlight = new Map< + string, + Promise<{ recomputed: boolean }> +>(); + +export async function ensureUserMoodRollupsFresh( + userId: string, +): Promise<{ recomputed: boolean }> { + const existing = ensureUserMoodRollupsFreshInFlight.get(userId); + if (existing) return existing; + const inflight = ensureUserMoodRollupsFreshImpl(userId).finally(() => { + ensureUserMoodRollupsFreshInFlight.delete(userId); + }); + ensureUserMoodRollupsFreshInFlight.set(userId, inflight); + return inflight; +} + +/** + * Test-only handle on the in-flight map. Mirrors the + * `_resetEnsureUserRollupsFreshInFlightForTests` pattern in + * `measurements/rollups.ts` — production code never calls this. + */ +export function _resetEnsureUserMoodRollupsFreshInFlightForTests(): void { + ensureUserMoodRollupsFreshInFlight.clear(); +} + +async function ensureUserMoodRollupsFreshImpl( + userId: string, +): Promise<{ recomputed: boolean }> { + try { + const [newestRollup, newestEntry] = await Promise.all([ + prisma.moodEntryRollup.findFirst({ + where: { userId, granularity: "DAY" }, + orderBy: { computedAt: "desc" }, + select: { computedAt: true }, + }), + prisma.moodEntry.findFirst({ + where: { userId }, + orderBy: { updatedAt: "desc" }, + select: { updatedAt: true, moodLoggedAt: true }, + }), + ]); + + if (!newestEntry) { + // No mood entries — rollups are trivially fresh. + return { recomputed: false }; + } + const watermark = + newestEntry.updatedAt > newestEntry.moodLoggedAt + ? newestEntry.updatedAt + : newestEntry.moodLoggedAt; + if (newestRollup && watermark <= newestRollup.computedAt) { + return { recomputed: false }; + } + + // Stale — full 5-year DAY fold. Bounded because mood entries are + // typically 1/day, so even a multi-year power user materialises a + // few thousand rows tops. + await recomputeUserMoodRollups(userId, { granularities: ["DAY"] }); + return { recomputed: true }; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + annotate({ + meta: { + mood_rollup_refresh_failed: true, + mood_rollup_refresh_error: message, + }, + }); + console.error("[mood-rollups] ensureUserMoodRollupsFresh failed:", message); + return { recomputed: false }; + } +} + +/** + * v1.4.39 — boot-time enqueue helper. Finds users with mood entries + * but zero `mood_entry_rollups` rows and enqueues one full-fold per + * account onto `MOOD_ROLLUP_FULL_BACKFILL_QUEUE`. Designed to be + * called once at worker boot so every self-hosted instance auto- + * converges on the persistent rollup tier without operator action. + * + * Idempotent across reboots: + * - the discovery query matches accounts with mood entries but + * zero rollup rows; once the fold completes the user drops off + * the list + * - pg-boss `singletonKey` coalesces duplicate sends within the + * queue, so a fast restart while a backfill is queued doesn't + * double up + * + * Best-effort: errors are swallowed and reported through the return + * value so the worker boot never fails because of a backfill miss. + */ +export async function enqueueBootTimeMoodRollupBackfill(): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + // Users with mood entries but no rollup coverage. Anchored on the + // per-user `DISTINCT user_id` set so the planner uses the + // `(user_id, date)` index path; the LEFT JOIN onto + // `mood_entry_rollups` uses the + // `(user_id, granularity, bucket_start)` composite primary key. + // `r."bucket_start" IS NULL` (any column from the right side of + // the LEFT JOIN) marks the unmatched users. + const users = await prisma.$queryRaw>` + SELECT DISTINCT mt."user_id" AS id + FROM ( + SELECT DISTINCT m."user_id" + FROM mood_entries m + ) mt + LEFT JOIN mood_entry_rollups r + ON r."user_id" = mt."user_id" + AND r."granularity" = 'DAY' + WHERE r."bucket_start" IS NULL + `; + + if (users.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { id } of users) { + const payload: MoodRollupFullBackfillPayload = { + userId: id, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(MOOD_ROLLUP_FULL_BACKFILL_QUEUE, payload, { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + // Coalesce: if a backfill for this user is already queued and + // we restart, pg-boss returns null instead of duplicating. + singletonKey: `mood-boot-backfill|${id}`, + }); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} + +// ─── internals ──────────────────────────────────────────── + +/** + * Pull the raw aggregates from Postgres. Groups by + * `date_trunc(, mood_logged_at)` and folds the stats + * with `STDDEV_POP`. The `date_trunc` unit literal must be inlined + * because Postgres rejects a parameterised expression in a GROUP BY; + * the value comes from a closed enum (DATE_TRUNC_UNIT) so there's no + * injection surface — we still assert against the whitelist before + * splicing. + */ +async function runMoodRollupAggregate( + client: PrismaTxOrClient, + input: { + userId: string; + granularity: RollupGranularity; + from: Date; + to: Date; + }, +): Promise { + const truncUnit = DATE_TRUNC_UNIT[input.granularity]; + if (!["day", "week", "month", "year"].includes(truncUnit)) { + throw new Error(`unexpected granularity: ${input.granularity}`); + } + const dateTrunc = `date_trunc('${truncUnit}', m."mood_logged_at")`; + + const sql = ` + SELECT + ${dateTrunc} AS bucket_start, + COUNT(*) 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" = $1 + AND m."mood_logged_at" >= $2 + AND m."mood_logged_at" < $3 + GROUP BY ${dateTrunc} + `; + return client.$queryRawUnsafe( + sql, + input.userId, + input.from, + input.to, + ); +} + +/** + * UPSERT the aggregate rows into `mood_entry_rollups`. Returns the + * number of rows touched. Chunked so a multi-year fold doesn't blow + * the parameter cap on a single round-trip. + */ +async function persistMoodRollupRows( + client: PrismaTxOrClient, + userId: string, + granularity: RollupGranularity, + rows: MoodRollupRow[], +): Promise { + if (rows.length === 0) return 0; + const CHUNK = 500; + 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, + }); + } + touched += slice.length; + } + return touched; +} + +/** + * Truncate a `Date` to the start of its UTC day. The rollup table + * stores TIMESTAMPTZ values so we anchor on UTC to keep the bucket + * key deterministic across timezone-shifting consumers — same + * convention as `measurement-rollups`. + */ +function startOfUtcDay(d: Date): Date { + return new Date( + Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate(), 0, 0, 0, 0), + ); +} + +/** + * Span of the bucket containing `moodLoggedAt` at the given + * granularity. Mirrors `measurement-rollups` so the WEEK / MONTH / + * YEAR fold queries select the same set of rows the read-side + * `date_trunc` would group on. + */ +function bucketSpan( + moodLoggedAt: Date, + granularity: RollupGranularity, +): { from: Date; to: Date } { + switch (granularity) { + case "DAY": { + const from = startOfUtcDay(moodLoggedAt); + return { from, to: new Date(from.getTime() + 24 * 60 * 60 * 1000) }; + } + case "WEEK": { + const day = startOfUtcDay(moodLoggedAt); + // Postgres ISO week: Monday is day 1. JS getUTCDay(): Sunday=0. + const dayOfWeek = day.getUTCDay(); + const mondayOffset = (dayOfWeek + 6) % 7; + const from = new Date( + day.getTime() - mondayOffset * 24 * 60 * 60 * 1000, + ); + return { from, to: new Date(from.getTime() + 7 * 24 * 60 * 60 * 1000) }; + } + case "MONTH": { + const from = new Date( + Date.UTC(moodLoggedAt.getUTCFullYear(), moodLoggedAt.getUTCMonth(), 1), + ); + const to = new Date( + Date.UTC( + moodLoggedAt.getUTCFullYear(), + moodLoggedAt.getUTCMonth() + 1, + 1, + ), + ); + return { from, to }; + } + case "YEAR": { + const from = new Date(Date.UTC(moodLoggedAt.getUTCFullYear(), 0, 1)); + const to = new Date(Date.UTC(moodLoggedAt.getUTCFullYear() + 1, 0, 1)); + return { from, to }; + } + } +}