diff --git a/src/lib/analytics/mood-prognosis/forecast.ts b/src/lib/analytics/mood-prognosis/forecast.ts new file mode 100644 index 000000000..e5cc768b1 --- /dev/null +++ b/src/lib/analytics/mood-prognosis/forecast.ts @@ -0,0 +1,148 @@ +/** + * One account's answer, computed and handed back as rows to write. + * + * Pure: it takes the days somebody logged and returns either the rows a pass + * should store or a refusal with the reason. It does not touch the database, + * and that is a boundary rather than a style — `MoodPrediction` has exactly + * one writer, the refresh job, and a guard holds it there + * (`src/__tests__/mood-prediction-write-surface-guard.test.ts`). Keeping the + * arithmetic here and the write there is what lets the whole decision path be + * tested without a container. + * + * **It refuses more often than it answers, and that is the design.** Too few + * days, no feature that clears the screening, no recent day carrying every + * feature the fit needs, a singular system — every one of them returns a + * refusal and the account is left with no forecast at all. The surface then + * says there is not enough to go on, which is true, instead of showing a + * number built out of three days and a coincidence. + */ +import { + buildFeatureMatrix, + standardiseDay, + type PrognosisDayInput, +} from "./features"; +import { PROGNOSIS_WRITE_WINDOW_DAYS } from "./load"; +import { featureContributions, predictWithFit } from "./ridge"; +import { forecastGate } from "./thresholds"; +import { fitPrognosisModel } from "./validate"; + +/** The lowest and highest value the target scale can express. */ +export const TARGET_MIN = 0; +export const TARGET_MAX = 10; + +/** One stored contribution: what a feature moved the expected value by. */ +export interface FeatureContribution { + feature: string; + contribution: number; +} + +/** One day's forecast, in the shape the row is written in. */ +export interface ForecastRow { + date: string; + predicted: number; + ciLow: number; + ciHigh: number; + n: number; + modelVersion: string; + /** Ordered by absolute contribution, largest first. */ + contributions: FeatureContribution[]; +} + +/** Why an account gets no forecast. Every one of these is a normal outcome. */ +export type ForecastRefusal = + /** Below the ladder's forecast threshold. */ + | "too-few-days" + /** Nothing recorded often enough, or varied enough, to screen in. */ + | "no-pattern" + /** A pattern, but no recent day carrying every feature it needs. */ + | "no-scorable-day"; + +export type ForecastResult = + | { + status: "written"; + rows: ForecastRow[]; + /** Complete days the fit used. The `n` every surface must show. */ + n: number; + modelVersion: string; + /** How wide the band came out, for the job's event only. */ + bandWidth: number; + featureCount: number; + } + | { status: "refused"; reason: ForecastRefusal }; + +function clampToScale(value: number): number { + return Math.min(TARGET_MAX, Math.max(TARGET_MIN, value)); +} + +function round(value: number, digits: number): number { + const factor = 10 ** digits; + return Math.round(value * factor) / factor; +} + +/** + * Fit the account's window and score its most recent days. + * + * The ladder is read against the number of DAYS carrying a self-rating, which + * is the number a person can see themselves growing. The `n` stored on each + * row is the number of complete days the fit actually used, which can be + * smaller — a day that is missing one of the fit's inputs cannot be part of + * it. Both numbers are honest and they answer different questions; the one on + * the surface beside the value is `n`, because that is what the value rests on. + */ +export function computeForecast( + days: readonly PrognosisDayInput[], + writeWindowDays: number = PROGNOSIS_WRITE_WINDOW_DAYS, +): ForecastResult { + const rated = days.filter((d) => typeof d.a1 === "number"); + const gate = forecastGate(rated.length); + if (!gate.present) return { status: "refused", reason: "too-few-days" }; + + const matrix = buildFeatureMatrix(days); + const model = matrix === null ? null : fitPrognosisModel(matrix); + if (model === null) return { status: "refused", reason: "no-pattern" }; + + const rows: ForecastRow[] = []; + for (const day of rated.slice(-writeWindowDays)) { + const standardised = standardiseDay(day, model.standardisation); + // A day missing one of the fit's features cannot be scored, and nothing is + // filled in for it. It simply has no forecast. + if (standardised === null) continue; + const raw = predictWithFit(model.fit, standardised); + if (raw === null) continue; + + rows.push({ + date: day.day, + predicted: round(clampToScale(raw), 2), + // The band travels with the prediction and is clamped to the same scale: + // a lower edge below zero would be a claim the scale cannot express. + ciLow: round(clampToScale(raw + model.bandLow), 2), + ciHigh: round(clampToScale(raw + model.bandHigh), 2), + n: model.n, + modelVersion: model.modelVersion, + contributions: featureContributions( + model.fit, + model.features, + standardised, + ) + .filter((c) => Number.isFinite(c.contribution)) + .sort((a, b) => Math.abs(b.contribution) - Math.abs(a.contribution)) + .map((c) => ({ + feature: c.feature, + contribution: round(c.contribution, 3), + })), + }); + } + + if (rows.length === 0) { + return { status: "refused", reason: "no-scorable-day" }; + } + + return { + status: "written", + rows, + n: model.n, + modelVersion: model.modelVersion, + bandWidth: round(model.bandHigh - model.bandLow, 2), + featureCount: model.features.length, + }; +} diff --git a/src/lib/analytics/mood-prognosis/load.ts b/src/lib/analytics/mood-prognosis/load.ts new file mode 100644 index 000000000..eb2ba1b99 --- /dev/null +++ b/src/lib/analytics/mood-prognosis/load.ts @@ -0,0 +1,190 @@ +/** + * The database read behind the fit, and nothing else. + * + * Separate from `features.ts` on purpose: that module is pure and is imported + * by the surface that labels a stored feature key, and a Prisma import in it + * would pull the database client towards a client bundle. The split is the + * boundary, not a preference. + * + * **One row per day, taken whole from one entry.** A day with two entries + * contributes the later one — its target, its dimensions and its context + * together. Averaging the ratings and taking the context from whichever entry + * happened to carry one would build a row nobody logged, and the model would + * then be fitted on days that did not happen. + * + * The day key is the entry's own `date`, already resolved in the zone the + * entry was written in. Re-deriving it here from `moodLoggedAt` would be a + * second opinion about which day a row belongs to, which is the class of bug + * the per-row `tz` column was added to end. + */ +import { prisma } from "@/lib/db"; +import { decodeKeyList } from "@/lib/mood/context-vocabulary"; +import { DEFAULT_TIMEZONE } from "@/lib/mood/date-key"; +import { resolveLinkedDayFigures } from "@/lib/mood/linked-context"; + +import type { PrognosisDayInput } from "./features"; + +/** + * How far back the fit looks. + * + * A year: long enough to hold a seasonal swing, short enough that a life that + * changed two years ago is not still being fitted against today. Longer would + * also make the per-user standardisation describe somebody the person no + * longer is. + */ +export const PROGNOSIS_FIT_WINDOW_DAYS = 365; + +/** + * How many recent days the job writes a forecast for. + * + * Not the whole window: a forecast is worth storing for the days somebody + * might still be looking at. Older rows are recomputable and are deleted on + * every pass, which is also what keeps a stale forecast from outliving the + * data it was fitted on. + */ +export const PROGNOSIS_WRITE_WINDOW_DAYS = 7; + +/** `YYYY-MM-DD`, this many days before the given instant, in UTC terms. */ +export function dayKeyBefore(now: Date, days: number): string { + return new Date(now.getTime() - days * 86_400_000).toISOString().slice(0, 10); +} + +/** + * Every day of the fit window, in day order, with everything the matrix needs. + * + * The linked figures come from the resolver that owns them, over the whole + * window in one query. They are read in the account's display zone: the entry + * rows carry their own zone for their own day key, but a step count has to be + * bucketed under some zone, and the account's own is the one every other + * surface uses. + */ +export async function loadPrognosisDays( + userId: string, + now: Date, + timezone: string | null, + windowDays: number = PROGNOSIS_FIT_WINDOW_DAYS, +): Promise { + const since = dayKeyBefore(now, windowDays); + const entries = await prisma.moodEntry.findMany({ + where: { userId, deletedAt: null, date: { gte: since } }, + // Ascending, so a later entry on the same day overwrites the earlier one + // in the fold below and the "last entry of the day wins" rule is the + // read order rather than a comparison. + orderBy: [{ date: "asc" }, { moodLoggedAt: "asc" }], + select: { + date: true, + moodA1: true, + stressA2: true, + energyA3: true, + connectionA4: true, + stabilityA5: true, + context: { + select: { + workStatus: true, + workMinutes: true, + overtimeMinutes: true, + workLoad: true, + workSatisfaction: true, + contactCircles: true, + contactForm: true, + contactExtent: true, + contactQuality: true, + contactSupport: true, + leisureCategories: true, + leisureMinutes: true, + leisureJoy: true, + leisureRecovery: true, + eventType: true, + eventValence: true, + // The note is deliberately not selected. It is encrypted free text + // about somebody's day and the model has no use for it; a fit does + // not read prose, and a query that decrypted it anyway would be + // reading a diary to compute an average. + }, + }, + }, + }); + + const byDay = new Map(); + for (const entry of entries) { + byDay.set(entry.date, { + day: entry.date, + a1: entry.moodA1, + dimensions: { + a2: entry.stressA2, + a3: entry.energyA3, + a4: entry.connectionA4, + a5: entry.stabilityA5, + }, + context: entry.context + ? { + workStatus: entry.context.workStatus, + workMinutes: entry.context.workMinutes, + overtimeMinutes: entry.context.overtimeMinutes, + workLoad: entry.context.workLoad, + workSatisfaction: entry.context.workSatisfaction, + contactCircles: decodeKeyList(entry.context.contactCircles), + contactForm: entry.context.contactForm, + contactExtent: entry.context.contactExtent, + contactQuality: entry.context.contactQuality, + contactSupport: entry.context.contactSupport, + leisureCategories: decodeKeyList(entry.context.leisureCategories), + leisureMinutes: entry.context.leisureMinutes, + leisureJoy: entry.context.leisureJoy, + leisureRecovery: entry.context.leisureRecovery, + eventType: entry.context.eventType, + eventValence: entry.context.eventValence, + } + : null, + linked: { + sleepAsleep: null, + steps: null, + activeEnergy: null, + restingHeartRate: null, + heartRateVariability: null, + }, + }); + } + + const days = [...byDay.keys()].sort(); + if (days.length === 0) return []; + + const linked = await resolveLinkedDayFigures( + userId, + days, + timezone ?? DEFAULT_TIMEZONE, + ); + for (const day of days) { + const figures = linked.get(day); + if (figures) byDay.get(day)!.linked = figures; + } + + return days.map((day) => byDay.get(day)!); +} + +/** + * How many days of the window carry a self-rated pleasantness. + * + * The number the ladder is read against, and the number a surface quotes when + * it explains why there is nothing yet. Days, not entries: two entries on one + * Tuesday are one Tuesday, and counting them as two would let somebody reach + * a threshold by logging twice a day for a fortnight. + */ +export async function countQualifyingDays( + userId: string, + now: Date, + windowDays: number = PROGNOSIS_FIT_WINDOW_DAYS, +): Promise { + const since = dayKeyBefore(now, windowDays); + const days = await prisma.moodEntry.findMany({ + where: { + userId, + deletedAt: null, + moodA1: { not: null }, + date: { gte: since }, + }, + distinct: ["date"], + select: { date: true }, + }); + return days.length; +} diff --git a/src/lib/jobs/__tests__/mood-prognosis-refresh.test.ts b/src/lib/jobs/__tests__/mood-prognosis-refresh.test.ts new file mode 100644 index 000000000..b3c77861d --- /dev/null +++ b/src/lib/jobs/__tests__/mood-prognosis-refresh.test.ts @@ -0,0 +1,152 @@ +/** + * The nightly mood-forecast rebuild. + * + * Three properties, each of which fails silently if it breaks: + * + * 1. **The sweep visits accounts that hold rows, not only accounts that + * qualify.** An account that deleted its entries drops out of the + * eligibility half of the cohort, and if that were the whole cohort it + * would keep last month's forecast forever, under whatever date the last + * pass stamped on it. Nothing on the surface would say so. + * 2. **A refusal clears.** Below the threshold, module off, no pattern — all + * three leave the account with no rows rather than with stale ones. + * 3. **One account's bad data costs one account.** The pass counts the + * failure and carries on; only an all-accounts failure fails the job. + * + * The cohort query and the module gate are mocked, which is what makes the + * decision path testable without a container. The arithmetic underneath has + * its own tests, and the whole thing is exercised end to end against real + * Postgres in `tests/integration/mood-prognosis-ladder.test.ts`. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/jobs/reminder/shared", () => ({ + getWorkerPrisma: () => ({ + moodEntry: { groupBy: groupByMock }, + moodPrediction: { + findMany: predictionFindManyMock, + deleteMany: deleteManyMock, + }, + }), +})); + +vi.mock("@/lib/modules/gate", () => ({ + isModuleEnabled: vi.fn(async () => true), +})); + +vi.mock("@/lib/logging/background", () => ({ + withBackgroundEvent: async ( + _name: string, + fn: (evt: { addMeta: () => void }) => Promise, + ) => fn({ addMeta: () => {} }), +})); + +const groupByMock = vi.fn(); +const predictionFindManyMock = vi.fn(); +const deleteManyMock = vi.fn(); + +import { + MOOD_PROGNOSIS_REFRESH_CRON, + MOOD_PROGNOSIS_REFRESH_QUEUE, + handleMoodPrognosisRefresh, + runMoodPrognosisRefresh, +} from "@/lib/jobs/mood-prognosis-refresh"; +import { MIN_ENTRIES_FOR_FORECAST } from "@/lib/analytics/mood-prognosis/thresholds"; + +const NOW = new Date("2026-08-08T02:35:00.000Z"); + +beforeEach(() => { + vi.clearAllMocks(); + groupByMock.mockResolvedValue([ + { userId: "user-1", _count: { _all: MIN_ENTRIES_FOR_FORECAST } }, + { userId: "user-2", _count: { _all: 120 } }, + ]); + predictionFindManyMock.mockResolvedValue([]); + deleteManyMock.mockResolvedValue({ count: 0 }); +}); + +describe("mood-prognosis-refresh", () => { + it("runs at 04:35, in the slot the neighbouring jobs leave free", () => { + expect(MOOD_PROGNOSIS_REFRESH_CRON).toBe("35 4 * * *"); + expect(MOOD_PROGNOSIS_REFRESH_QUEUE).toBe("mood-prognosis-refresh"); + }); + + it("refreshes every eligible account", async () => { + const refresh = vi.fn(async () => ({ rows: 7, refusal: null })); + const summary = await runMoodPrognosisRefresh(NOW, { refresh }); + + expect(summary.scanned).toBe(2); + // The count is the point: a cohort query that matched nothing would + // satisfy "no bad rows" exactly as well as a working sweep. + expect(refresh).toHaveBeenCalledTimes(2); + expect(refresh).toHaveBeenCalledWith("user-1", NOW); + expect(summary.refreshed).toBe(2); + expect(summary.rowsWritten).toBe(14); + }); + + it("leaves an account below the entry cut out of the cohort", async () => { + groupByMock.mockResolvedValue([ + { userId: "user-1", _count: { _all: MIN_ENTRIES_FOR_FORECAST - 1 } }, + ]); + const refresh = vi.fn(async () => ({ rows: 1, refusal: null })); + const summary = await runMoodPrognosisRefresh(NOW, { refresh }); + + expect(summary.scanned).toBe(0); + expect(refresh).not.toHaveBeenCalled(); + }); + + it("still visits an account that holds rows but no longer qualifies", async () => { + groupByMock.mockResolvedValue([]); + predictionFindManyMock.mockResolvedValue([{ userId: "user-stale" }]); + const refresh = vi.fn(async () => ({ + rows: 0, + refusal: "too-few-days" as const, + })); + + const summary = await runMoodPrognosisRefresh(NOW, { refresh }); + + // Without this arm the account would keep its forecast forever: it is not + // in the eligible set any more, so nothing would ever come back for it. + expect(summary.scanned).toBe(1); + expect(refresh).toHaveBeenCalledWith("user-stale", NOW); + expect(summary.refused).toBe(1); + expect(summary.refreshed).toBe(0); + }); + + it("computes nothing and stores nothing while the module is off, and clears what it stored before", async () => { + const refresh = vi.fn(async () => ({ rows: 7, refusal: null })); + const summary = await runMoodPrognosisRefresh(NOW, { + isAvailable: async () => false, + refresh, + }); + + expect(refresh).not.toHaveBeenCalled(); + expect(summary.moduleOff).toBe(2); + expect(summary.rowsWritten).toBe(0); + expect(deleteManyMock).toHaveBeenCalledWith({ + where: { userId: "user-1" }, + }); + expect(deleteManyMock).toHaveBeenCalledTimes(2); + }); + + it("counts an account whose pass throws and carries on", async () => { + const refresh = vi + .fn() + .mockRejectedValueOnce(new Error("bad rows")) + .mockResolvedValueOnce({ rows: 3, refusal: null }); + + const summary = await runMoodPrognosisRefresh(NOW, { refresh }); + + expect(summary.failed).toBe(1); + expect(summary.refreshed).toBe(1); + expect(summary.rowsWritten).toBe(3); + }); + + it("fails the job only when every account failed", async () => { + const outcome = await handleMoodPrognosisRefresh([]); + // The default refresh path reaches the real loader, which the mocked + // client cannot serve, so every account throws — which is exactly the + // all-accounts case. + expect(outcome.ok).toBe(false); + }); +}); diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 2c84daf4b..2c255d692 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -121,6 +121,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "medications_evaluated", "metric_assessments_warmed", "missed_doses_minted", + "module_off", "mood_entries_migrated", "mood_pruned", "notifications_dispatched", @@ -147,8 +148,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "rearmed", "records_read", "refreshed", + "refused", "reminders_due", "removed", + "rows_written", "restore_drill_age_days", "restore_drill_ciphertext_bytes", "restore_drill_intake_events", diff --git a/src/lib/jobs/mood-prognosis-refresh.ts b/src/lib/jobs/mood-prognosis-refresh.ts new file mode 100644 index 000000000..30ab95fba --- /dev/null +++ b/src/lib/jobs/mood-prognosis-refresh.ts @@ -0,0 +1,241 @@ +/** + * Nightly rebuild of the mood forecast rows. + * + * **This is not an AI surface.** A ridge regression over a dozen standardised + * columns is arithmetic: no provider, no key, no egress, no consent gate. It + * is stated here because the shape of the feature invites the assumption in + * both directions — nobody should wire a consent check that does not apply, + * and nobody should assume one exists where none is needed. If a later change + * narrates this forecast through a language model, THAT call site inherits + * `ai-consent-enforcement-guard.test.ts` and has to declare its surface. + * + * **This file is the only writer of `MoodPrediction`**, held there by + * `src/__tests__/mood-prediction-write-surface-guard.test.ts`. The reason is + * the product rule: a day carries the person's own rating and the value their + * past days imply, the two are never merged, and the second one is not + * editable. A route that wrote this table would make it editable. The cycle + * module learned the other half of the same lesson the expensive way — a + * forecast written as a side effect of opening a page is a function of + * browsing rather than of data. + * + * **A pass replaces; it never accumulates.** Every account it touches ends the + * pass holding exactly the rows this run wrote, or none. An account that has + * fallen below the threshold, switched the module off, or stopped showing a + * pattern is left with nothing rather than with last week's answer under a + * fresh-looking date. That is the stale rule, chosen over marking rows + * superseded because a derived row is cheap to rebuild and a superseded row is + * one more state every reader has to handle correctly. + * + * Runs at 04:35 Europe/Berlin. The 03:xx block is retention; 04:10 is the + * document purge, 04:20 the cycle forecast, 04:25 the achievement sweep, 04:30 + * the insight pre-generation, and 04:45/50/55 the recovery, stress and strain + * scores. 04:35 is free, and it sits after the inputs this job reads have + * settled. (04:40 is not free: the geo backfill runs at `40 * * * *`.) + */ +import type { Job } from "pg-boss"; + +import { + computeForecast, + type ForecastRefusal, +} from "@/lib/analytics/mood-prognosis/forecast"; +import { + PROGNOSIS_FIT_WINDOW_DAYS, + dayKeyBefore, + loadPrognosisDays, +} from "@/lib/analytics/mood-prognosis/load"; +import { MIN_ENTRIES_FOR_FORECAST } from "@/lib/analytics/mood-prognosis/thresholds"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { isModuleEnabled } from "@/lib/modules/gate"; + +import { getWorkerPrisma } from "./reminder/shared"; + +export const MOOD_PROGNOSIS_REFRESH_QUEUE = "mood-prognosis-refresh"; + +export const MOOD_PROGNOSIS_REFRESH_CRON = "35 4 * * *"; + +export interface MoodPrognosisRefreshPayload { + triggeredAt: string; +} + +export interface MoodPrognosisRefreshSummary { + /** Accounts the cohort query returned, plus any holding stale rows. */ + scanned: number; + /** Accounts whose forecast rows now reflect their data. */ + refreshed: number; + /** Rows written across every account. */ + rowsWritten: number; + /** Accounts the fit refused, whose rows were cleared. */ + refused: number; + /** Accounts whose module is off, whose rows were cleared. */ + moduleOff: number; + /** Accounts whose pass threw. */ + failed: number; +} + +/** + * Rebuild one account's rows, or clear them. + * + * The write and the delete are the two calls the guard allows in this file, + * and they are here rather than beside the arithmetic so the single-writer + * claim is a property of one module rather than a convention across two. + */ +async function refreshAccount( + userId: string, + now: Date, +): Promise<{ rows: number; refusal: ForecastRefusal | null }> { + const prisma = getWorkerPrisma(); + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { timezone: true }, + }); + const days = await loadPrognosisDays(userId, now, user?.timezone ?? null); + const result = computeForecast(days); + + if (result.status === "refused") { + await prisma.moodPrediction.deleteMany({ where: { userId } }); + return { rows: 0, refusal: result.reason }; + } + + const written: string[] = []; + for (const row of result.rows) { + const data = { + predicted: row.predicted, + ciLow: row.ciLow, + ciHigh: row.ciHigh, + n: row.n, + modelVersion: row.modelVersion, + features: JSON.stringify(row.contributions), + computedAt: now, + }; + await prisma.moodPrediction.upsert({ + where: { userId_date: { userId, date: row.date } }, + create: { userId, date: row.date, ...data }, + update: data, + }); + written.push(row.date); + } + // Everything outside this pass goes, so no account holds a row from an older + // fit beside a row from this one. + await prisma.moodPrediction.deleteMany({ + where: { userId, date: { notIn: written } }, + }); + return { rows: written.length, refusal: null }; +} + +/** + * Refresh every eligible account. Exported for the test; the handler below is + * the queue binding. + * + * The cohort is the union of two sets, and the second one is the whole point: + * accounts with enough recent rated days to be worth fitting, AND accounts + * that already hold forecast rows. Without the second, an account that deleted + * its entries would drop out of the cohort and keep its old forecast forever — + * the sweep would never visit it again to clear it. + */ +export async function runMoodPrognosisRefresh( + now: Date = new Date(), + options: { + isAvailable?: (userId: string) => Promise; + refresh?: typeof refreshAccount; + } = {}, +): Promise { + const isAvailable = + options.isAvailable ?? + ((userId: string) => isModuleEnabled(userId, "mood")); + const refresh = options.refresh ?? refreshAccount; + const prisma = getWorkerPrisma(); + + const since = dayKeyBefore(now, PROGNOSIS_FIT_WINDOW_DAYS); + const [eligible, holders] = await Promise.all([ + prisma.moodEntry.groupBy({ + by: ["userId"], + where: { + deletedAt: null, + moodA1: { not: null }, + date: { gte: since }, + }, + _count: { _all: true }, + }), + prisma.moodPrediction.findMany({ + distinct: ["userId"], + select: { userId: true }, + }), + ]); + + const userIds = new Set(); + for (const group of eligible) { + // The cheap cut, so an account with a handful of entries never pays for a + // window read. The real gate is the ladder inside the fit, which counts + // DAYS rather than rows. + if (group._count._all >= MIN_ENTRIES_FOR_FORECAST) + userIds.add(group.userId); + } + for (const holder of holders) userIds.add(holder.userId); + + const summary: MoodPrognosisRefreshSummary = { + scanned: userIds.size, + refreshed: 0, + rowsWritten: 0, + refused: 0, + moduleOff: 0, + failed: 0, + }; + + for (const userId of userIds) { + try { + if (!(await isAvailable(userId))) { + // An operator- or user-disabled module computes nothing and stores + // nothing — and gives back anything it stored before. + await prisma.moodPrediction.deleteMany({ where: { userId } }); + summary.moduleOff += 1; + continue; + } + const outcome = await refresh(userId, now); + if (outcome.refusal !== null) { + summary.refused += 1; + continue; + } + summary.refreshed += 1; + summary.rowsWritten += outcome.rows; + } catch { + // One account's bad data must not cost every other account its forecast. + // The count is what makes a systemic failure visible. + summary.failed += 1; + } + } + + return summary; +} + +export async function handleMoodPrognosisRefresh( + jobs: Job[], +): Promise { + void jobs; + return withBackgroundEvent("job.mood-prognosis-refresh", async (evt) => { + const summary = await runMoodPrognosisRefresh(); + evt.addMeta("mood_prognosis_scanned", summary.scanned); + evt.addMeta("mood_prognosis_refreshed", summary.refreshed); + evt.addMeta("mood_prognosis_rows", summary.rowsWritten); + evt.addMeta("mood_prognosis_refused", summary.refused); + evt.addMeta("mood_prognosis_module_off", summary.moduleOff); + evt.addMeta("mood_prognosis_failed", summary.failed); + + // Every account failing is a systemic fault (a dead pool, a schema drift), + // not one account's bad data. Anything short of that stays a success with + // the count on the event. + if (summary.failed > 0 && summary.failed === summary.scanned) { + return jobFailed( + `mood-prognosis-refresh failed for all ${summary.failed} accounts`, + ); + } + return jobDone({ + users_scanned: summary.scanned, + refreshed: summary.refreshed, + rows_written: summary.rowsWritten, + refused: summary.refused, + module_off: summary.moduleOff, + users_failed: summary.failed, + }); + }); +} diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 332dd8c1d..6eee62d3b 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -195,6 +195,12 @@ import { handleCyclePredictionRefresh, type CyclePredictionRefreshPayload, } from "@/lib/jobs/cycle-prediction-refresh"; +import { + MOOD_PROGNOSIS_REFRESH_QUEUE, + MOOD_PROGNOSIS_REFRESH_CRON, + handleMoodPrognosisRefresh, + type MoodPrognosisRefreshPayload, +} from "@/lib/jobs/mood-prognosis-refresh"; import { ACHIEVEMENT_UNLOCK_SWEEP_QUEUE, ACHIEVEMENT_UNLOCK_SWEEP_CRON, @@ -365,6 +371,10 @@ const allQueues = [ // scans. Without this entry the schedule silently no-ops and the forecast // is only as fresh as the last time somebody opened the calendar. CYCLE_PREDICTION_REFRESH_QUEUE, + // v1.38 — nightly rebuild of the mood forecast rows. Without this entry the + // schedule silently no-ops: nothing else writes them, so the surface would + // answer "not enough data yet" forever on an account that has plenty. + MOOD_PROGNOSIS_REFRESH_QUEUE, // v1.35.3 — nightly pin for newly unlocked achievements. Without this entry // the schedule silently no-ops and no unlock date is ever made durable. ACHIEVEMENT_UNLOCK_SWEEP_QUEUE, @@ -521,6 +531,10 @@ const schedules: ScheduleEntry[] = [ CYCLE_PREDICTION_REFRESH_CRON, cronIsTheRetry, ], + // v1.38 — daily 04:35 Europe/Berlin rebuild of the mood forecast rows. + // Tomorrow's tick is the retry: the fit moves on a day scale and a night + // late writes the same rows from the same days. + [MOOD_PROGNOSIS_REFRESH_QUEUE, MOOD_PROGNOSIS_REFRESH_CRON, cronIsTheRetry], // v1.35.3 — daily 04:25 Europe/Berlin pin for newly unlocked achievements. // Tomorrow's tick is the retry: the date written is the computed completion // date, so a night late writes the identical row. @@ -743,6 +757,14 @@ export async function registerMaintenanceQueues( { localConcurrency: 1 }, handleCyclePredictionRefresh, ); + // v1.38 — nightly mood-forecast rebuild. Single-flight: two ticks would fit + // the same windows twice and the second pass would write the identical rows. + await createAndWork( + boss, + MOOD_PROGNOSIS_REFRESH_QUEUE, + { localConcurrency: 1 }, + handleMoodPrognosisRefresh, + ); // v1.35.3 — nightly achievement-unlock pin. Single-flight: the underlying // createMany is idempotent on the (userId, achievementId) unique, so a // second concurrent tick would only re-walk the same accounts. diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index 04550e269..89dd1f05f 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -117,6 +117,7 @@ const subsystemSurface = { "push-attempt-cleanup": { audience: "system" }, "arrival-reaction-cleanup": { audience: "system" }, "cycle-prediction-refresh": { audience: "system" }, + "mood-prognosis-refresh": { audience: "system" }, "achievement-unlock-sweep": { audience: "system" }, "measurement-tombstone-cleanup": { audience: "system" }, "coach-message-cleanup": { audience: "system" },