diff --git a/src/lib/jobs/__tests__/cumulative-pr-rederive-queue.test.ts b/src/lib/jobs/__tests__/cumulative-pr-rederive-queue.test.ts new file mode 100644 index 000000000..61aa8c699 --- /dev/null +++ b/src/lib/jobs/__tests__/cumulative-pr-rederive-queue.test.ts @@ -0,0 +1,48 @@ +/** + * v1.30.3 (QA F4) — cumulative-PersonalRecord rederivation queue + * registration guard. + * + * Same source-text-grep approach as the step-consolidation-repair guard: + * assert the queue is registered in `allQueues`, a `boss.work` handler is + * wired against it, and the boot-discovery enqueue helper fires — without + * booting pg-boss + Prisma. An unregistered queue silently never drains + * (the v1.4.37 dead-queue class). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +const REGISTRAR_PATH = join(__dirname, "..", "reminder", "register-rollup.ts"); +const source = readFileSync(REGISTRAR_PATH, "utf8"); + +describe("reminder-worker — cumulative-pr-rederive wiring", () => { + it("imports the queue symbols from the rederivation module", () => { + expect(source).toMatch( + /from\s*["']@\/lib\/personal-records\/cumulative-pr-rederivation["']/, + ); + expect(source).toMatch(/\bCUMULATIVE_PR_REDERIVE_QUEUE\b/); + expect(source).toMatch(/\brunCumulativePrRederivationForUser\b/); + expect(source).toMatch(/\benqueueBootTimeCumulativePrRederivation\b/); + }); + + it("registers the queue in the allQueues loop", () => { + const allQueuesMatch = source.match( + /const allQueues\s*=\s*\[([\s\S]*?)\];/, + ); + expect(allQueuesMatch).not.toBeNull(); + expect(allQueuesMatch![1]).toMatch(/\bCUMULATIVE_PR_REDERIVE_QUEUE\b/); + }); + + it("registers a boss.work handler against the queue", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,200}CUMULATIVE_PR_REDERIVE_QUEUE[\s\S]{0,400}runCumulativePrRederivationForUser/, + ); + }); + + it("fires the boot-discovery enqueue helper", () => { + expect(source).toMatch( + /await enqueueBootTimeCumulativePrRederivation\([^)]*\)/, + ); + }); +}); diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index ded392022..ce03e656c 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -67,6 +67,13 @@ import { enqueueBootTimeStepConsolidationRepair, type StepConsolidationRepairPayload, } from "@/lib/jobs/step-consolidation-repair"; +import { + CUMULATIVE_PR_REDERIVE_QUEUE, + CUMULATIVE_PR_REDERIVE_CONCURRENCY, + runCumulativePrRederivationForUser, + enqueueBootTimeCumulativePrRederivation, + type CumulativePrRederivePayload, +} from "@/lib/personal-records/cumulative-pr-rederivation"; import { MEAN_CONSOLIDATION_QUEUE, MEAN_CONSOLIDATION_CONCURRENCY, @@ -132,6 +139,10 @@ const allQueues = [ // v1.28.37 — one-shot repair for the provider step rows the pre-fix // consolidation swept. Boot-discovery driven, self-converging. STEP_CONSOLIDATION_REPAIR_QUEUE, + // v1.30.3 — one-shot repair for the cumulative-type PersonalRecord rows + // the pre-fix multi-source SUM inflated. Boot-discovery driven, + // self-converging (QA F4). + CUMULATIVE_PR_REDERIVE_QUEUE, // v1.7.0 — daily-mean consolidation for high-frequency spot HealthKit // metrics (walking speed/step length, respiratory rate, audio exposure). MEAN_CONSOLIDATION_QUEUE, @@ -248,6 +259,27 @@ export async function registerRollupQueues( }, ); + // v1.30.3 — cumulative-PersonalRecord rederivation worker (QA F4). The + // boot enqueue helper below sends one job per user still holding a + // suspect pre-fix inflated cumulative-type record; this handler deletes + // it and re-runs detection silently so the honest re-derived best takes + // its place. Serial concurrency so the repair never crowds the + // dashboard request pool. + await boss.work( + CUMULATIVE_PR_REDERIVE_QUEUE, + { localConcurrency: CUMULATIVE_PR_REDERIVE_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + const summary = await runCumulativePrRederivationForUser(userId); + workerLog( + "info", + `[cumulative-pr-rederive] user=${userId} rowsDeleted=${summary.rowsDeleted} rowsReinserted=${summary.rowsReinserted}`, + ); + } + }, + ); + // v1.7.0 — daily-mean consolidation worker. The boot enqueue helper // below sends one job per user holding live per-sample high-frequency // mean-type rows; this handler collapses each completed day to its @@ -771,4 +803,35 @@ export async function enqueueRollupBootDiscovery(): Promise { err, ); } + + // v1.30.3 (QA F4) — cumulative-PersonalRecord rederivation. Finds every + // account still holding a suspect measurement-driven cumulative-type + // record created before the source-collapse fix landed and enqueues one + // repair job per account. Self-converging: a deleted-and-redetected + // record's `createdAt` sits after the fix cutoff, so the account drops + // out of the discovery predicate for good. + try { + // Stage 6. + const { enqueued, skipped, error } = + await enqueueBootTimeCumulativePrRederivation( + BOOT_BACKFILL_STAGGER_SECONDS * 6, + ); + if (error) { + workerLog( + "error", + `[cumulative-pr-rederive] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[cumulative-pr-rederive] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[cumulative-pr-rederive] boot discovery threw an unexpected error", + err, + ); + } } diff --git a/src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts b/src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts new file mode 100644 index 000000000..e4172f434 --- /dev/null +++ b/src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts @@ -0,0 +1,166 @@ +/** + * v1.30.3 (QA F4) — the one-shot repair for the pre-fix inflated + * cumulative-type PersonalRecord rows: deletes the suspect row(s) for a + * user and re-runs detection silently so the honest re-derived best takes + * its place, plus the boot-discovery enqueue that finds affected accounts. + */ +import { describe, expect, it, vi, beforeEach } from "vitest"; + +const deleteMany = vi.fn(); +const findMany = vi.fn(); +vi.mock("@/lib/db", () => ({ + prisma: { + personalRecord: { + deleteMany: (a: unknown) => deleteMany(a), + findMany: (a: unknown) => findMany(a), + }, + }, +})); + +const detectPersonalRecordsForUser = vi.fn(); +vi.mock("@/lib/personal-records/pr-detection-worker", () => ({ + detectPersonalRecordsForUser: (...args: unknown[]) => + detectPersonalRecordsForUser(...args), +})); + +const send = vi.fn(); +const getGlobalBoss = vi.fn(); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => getGlobalBoss(), +})); + +const annotate = vi.fn(); +vi.mock("@/lib/logging/context", () => ({ + annotate: (...a: unknown[]) => annotate(...a), +})); + +import { + runCumulativePrRederivationForUser, + enqueueBootTimeCumulativePrRederivation, + CUMULATIVE_PR_FIX_CUTOFF, +} from "@/lib/personal-records/cumulative-pr-rederivation"; + +describe("runCumulativePrRederivationForUser", () => { + beforeEach(() => { + deleteMany.mockReset(); + detectPersonalRecordsForUser.mockReset(); + annotate.mockReset(); + }); + + it("deletes suspect rows created before the fix cutoff and re-derives silently", async () => { + deleteMany.mockResolvedValue({ count: 1 }); + detectPersonalRecordsForUser.mockResolvedValue({ + inserted: 1, + ties: 0, + scanned: 6, + silent: true, + }); + + const summary = await runCumulativePrRederivationForUser("u1"); + + expect(deleteMany).toHaveBeenCalledWith({ + where: { + userId: "u1", + metricSlot: null, + metricType: { in: expect.arrayContaining(["ACTIVITY_STEPS"]) }, + createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF }, + }, + }); + // Re-derivation always runs silent — this is data hygiene, not a new + // achievement, so no push should fire for it. + expect(detectPersonalRecordsForUser).toHaveBeenCalledWith("u1", { + silent: true, + }); + expect(summary).toEqual({ rowsDeleted: 1, rowsReinserted: 1 }); + }); + + it("skips re-derivation entirely when nothing was deleted", async () => { + deleteMany.mockResolvedValue({ count: 0 }); + + const summary = await runCumulativePrRederivationForUser("u2"); + + expect(detectPersonalRecordsForUser).not.toHaveBeenCalled(); + expect(summary).toEqual({ rowsDeleted: 0, rowsReinserted: 0 }); + }); + + it("counts a tie as a reinserted row (the honest re-derived value can equal a still-live sibling metric's best)", async () => { + deleteMany.mockResolvedValue({ count: 2 }); + detectPersonalRecordsForUser.mockResolvedValue({ + inserted: 0, + ties: 1, + scanned: 6, + silent: true, + }); + + const summary = await runCumulativePrRederivationForUser("u3"); + expect(summary).toEqual({ rowsDeleted: 2, rowsReinserted: 1 }); + }); +}); + +describe("enqueueBootTimeCumulativePrRederivation", () => { + beforeEach(() => { + findMany.mockReset(); + send.mockReset(); + getGlobalBoss.mockReset(); + }); + + it("no-ops when no boss is attached (test / scriptless context)", async () => { + getGlobalBoss.mockReturnValue(null); + const result = await enqueueBootTimeCumulativePrRederivation(); + expect(result).toEqual({ enqueued: 0, skipped: 0, error: null }); + expect(findMany).not.toHaveBeenCalled(); + }); + + it("finds zero affected users and enqueues nothing", async () => { + getGlobalBoss.mockReturnValue({ send }); + findMany.mockResolvedValue([]); + const result = await enqueueBootTimeCumulativePrRederivation(); + expect(result).toEqual({ enqueued: 0, skipped: 0, error: null }); + expect(send).not.toHaveBeenCalled(); + }); + + it("enqueues one job per distinct affected user with a coalescing singletonKey", async () => { + getGlobalBoss.mockReturnValue({ send }); + findMany.mockResolvedValue([{ userId: "u1" }, { userId: "u2" }]); + send.mockResolvedValue("job-id"); + + const result = await enqueueBootTimeCumulativePrRederivation(30); + + expect(result).toEqual({ enqueued: 2, skipped: 0, error: null }); + expect(findMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + metricSlot: null, + createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF }, + }), + distinct: ["userId"], + }), + ); + expect(send).toHaveBeenCalledWith( + "cumulative-pr-rederive", + expect.objectContaining({ userId: "u1" }), + expect.objectContaining({ + singletonKey: "cumulative-pr-rederive|u1", + startAfter: 30, + }), + ); + }); + + it("counts a coalesced send (pg-boss returns null) as skipped, not an error", async () => { + getGlobalBoss.mockReturnValue({ send }); + findMany.mockResolvedValue([{ userId: "u1" }]); + send.mockResolvedValue(null); + + const result = await enqueueBootTimeCumulativePrRederivation(); + expect(result).toEqual({ enqueued: 0, skipped: 1, error: null }); + }); + + it("returns the error message instead of throwing on a discovery failure", async () => { + getGlobalBoss.mockReturnValue({ send }); + findMany.mockRejectedValue(new Error("db down")); + + const result = await enqueueBootTimeCumulativePrRederivation(); + expect(result.enqueued).toBe(0); + expect(result.error).toBe("db down"); + }); +}); diff --git a/src/lib/personal-records/cumulative-pr-rederivation.ts b/src/lib/personal-records/cumulative-pr-rederivation.ts new file mode 100644 index 000000000..a6202ac24 --- /dev/null +++ b/src/lib/personal-records/cumulative-pr-rederivation.ts @@ -0,0 +1,184 @@ +/** + * v1.30.3 (QA F4) — one-shot repair for the `PersonalRecord` rows the + * pre-fix cumulative-day picker double-counted. + * + * Background: before `8902d3985` (v1.29.6), `findBestCumulativeDay` + * (`pr-detection-worker.ts`) summed EVERY row on a calendar day regardless + * of source — a day with both an Apple Health AND a Withings write for the + * same cumulative metric (steps, active energy, flights, distance, + * daylight, falls) summed both sources into one inflated total and wrote + * it as the all-time "best". The fix (source-collapse via + * `pickCanonicalSourceRows` before summing) stops the inflation going + * forward, but the worker's improvement gate only ever compares a + * freshly-computed day against the STORED best — an already-inflated + * stored row can never be beaten again, so the false record silently + * outranks every honest day forever (PR list, digest milestone, doctor + * report), and the affected user can never set a genuine record on that + * metric again. + * + * This job deletes the suspect rows (measurement-driven — `metricSlot` + * null — one of `CUMULATIVE_HK_TYPES`, created BEFORE the fix landed) and + * immediately re-runs detection for the affected user with `silent: true` + * so the (already-fixed) worker inserts the honest re-derived best without + * firing a "new record" push for what is data hygiene, not an achievement. + * + * Converges to zero without a marker column: the cutoff is the fixed + * instant the source-collapse fix went live upstream (safely after the + * `8902d3985` merge). Any row created before it predates the bug fix and + * is suspect; any row this job (or the ordinary worker) writes going + * forward is always created after that instant, so it can never re-match + * the discovery predicate. A re-run (reboot mid-walk, or an install that + * upgrades straight past this release) simply finds fewer/no affected + * users each time. + * + * The queue name MUST be registered in `allQueues` in + * `src/lib/jobs/reminder/register-rollup.ts` or pg-boss never provisions + * it and the boot enqueue silently never drains (the v1.4.37 dead-queue + * class). + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { detectPersonalRecordsForUser } from "@/lib/personal-records/pr-detection-worker"; +import type { MeasurementType } from "@/generated/prisma/client"; + +export const CUMULATIVE_PR_REDERIVE_QUEUE = "cumulative-pr-rederive"; + +/** + * Serial concurrency — a repair walks one account's PersonalRecord rows + * and re-runs full detection; matches the other one-shot repair jobs + * (`STEP_CONSOLIDATION_REPAIR_CONCURRENCY`). + */ +export const CUMULATIVE_PR_REDERIVE_CONCURRENCY = 1; + +/** + * The instant the cumulative-day source-collapse fix (`8902d3985`, + * v1.29.6) went live. Any measurement-driven `PersonalRecord` row for a + * cumulative type `createdAt` STRICTLY BEFORE this instant was written by + * the pre-fix multi-source SUM and is suspect. Hardcoded (not derived from + * git/commit metadata at runtime) so the predicate is a compile-time + * constant every self-hosted install evaluates identically regardless of + * when it upgrades to this release. + */ +export const CUMULATIVE_PR_FIX_CUTOFF = new Date("2026-07-18T00:00:00.000Z"); + +const CUMULATIVE_TYPES_ARRAY: MeasurementType[] = [ + ...CUMULATIVE_HK_TYPES, +] as MeasurementType[]; + +export interface CumulativePrRederivePayload { + userId: string; + enqueuedAt: string; +} + +export interface CumulativePrRederiveSummary { + /** Suspect rows deleted for this user. */ + rowsDeleted: number; + /** Records the (already-fixed) worker re-inserted after the delete. */ + rowsReinserted: number; +} + +/** + * Per-user repair handler. Deletes the suspect measurement-driven + * cumulative-type `PersonalRecord` rows for `userId`, then re-runs + * detection silently so the honest re-derived best (if any — the warm-up + * gate can still apply) replaces it. Safe to re-run: a second pass finds + * zero suspect rows once the honest one has replaced them (its + * `createdAt` is always after the cutoff). + */ +export async function runCumulativePrRederivationForUser( + userId: string, +): Promise { + const { count: rowsDeleted } = await prisma.personalRecord.deleteMany({ + where: { + userId, + metricSlot: null, + metricType: { in: CUMULATIVE_TYPES_ARRAY }, + createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF }, + }, + }); + + let rowsReinserted = 0; + if (rowsDeleted > 0) { + const result = await detectPersonalRecordsForUser(userId, { + silent: true, + }); + rowsReinserted = result.inserted + result.ties; + } + + annotate({ + action: { + name: "personal-record.cumulative.rederive", + details: { rows_deleted: rowsDeleted, rows_reinserted: rowsReinserted }, + }, + }); + + return { rowsDeleted, rowsReinserted }; +} + +/** + * Boot-time discovery. Finds every account still holding a suspect + * measurement-driven cumulative-type `PersonalRecord` row (created before + * the fix cutoff) and enqueues one repair job per account. + * + * Self-converging: once a suspect row is deleted and (when the warm-up + * gate clears) replaced by an honestly re-derived one, its `createdAt` + * sits after the cutoff and the account drops off the discovery list for + * good. Best-effort: errors return through the result so worker boot never + * fails on a repair miss. + */ +export async function enqueueBootTimeCumulativePrRederivation( + startAfterSeconds: number = 0, +): Promise<{ enqueued: number; skipped: number; error: string | null }> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + const affected = await prisma.personalRecord.findMany({ + where: { + metricSlot: null, + metricType: { in: CUMULATIVE_TYPES_ARRAY }, + createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF }, + }, + distinct: ["userId"], + select: { userId: true }, + }); + + if (affected.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { userId } of affected) { + const payload: CumulativePrRederivePayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(CUMULATIVE_PR_REDERIVE_QUEUE, payload, { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + // Coalesce: if a repair for this user is already queued and we + // restart, pg-boss returns null instead of duplicating. + singletonKey: `cumulative-pr-rederive|${userId}`, + ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), + }); + 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), + }; + } +}