diff --git a/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts b/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts new file mode 100644 index 000000000..4daea0d98 --- /dev/null +++ b/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts @@ -0,0 +1,101 @@ +/** + * v1.10.0 — computed scores (WX-E). Queue-registration guards for the + * Stress + Strain nightly score jobs and the dense intra-day retention + * drain. + * + * Same source-text-grep approach as the WX-C recovery-score guard: assert + * each queue is imported, registered in `allQueues`, scheduled (the two + * cron-driven score jobs), and wired to a `boss.work` handler — without + * booting pg-boss + Prisma. An unregistered queue silently never drains + * (the recurring past bug this guards against). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts"); +const source = readFileSync(REMINDER_WORKER_PATH, "utf8"); + +function allQueuesBlock(): string { + const m = source.match(/const allQueues\s*=\s*\[([\s\S]*?)\];/); + expect(m).not.toBeNull(); + return m![1]; +} + +describe("reminder-worker — stress-score wiring", () => { + it("imports the queue symbols from the stress-score module", () => { + expect(source).toMatch(/from\s*["']@\/lib\/jobs\/stress-score["']/); + expect(source).toMatch(/\bSTRESS_SCORE_QUEUE\b/); + expect(source).toMatch(/\bSTRESS_SCORE_CRON\b/); + expect(source).toMatch(/\brunStressScore\b/); + }); + + it("registers the stress-score queue in allQueues", () => { + expect(allQueuesBlock()).toMatch(/\bSTRESS_SCORE_QUEUE\b/); + }); + + it("schedules the stress-score cron", () => { + expect(source).toMatch(/\[STRESS_SCORE_QUEUE,\s*STRESS_SCORE_CRON\]/); + }); + + it("registers a boss.work handler that runs the stress-score pass", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,200}STRESS_SCORE_QUEUE[\s\S]{0,400}runStressScore/, + ); + }); +}); + +describe("reminder-worker — strain-score wiring", () => { + it("imports the queue symbols from the strain-score module", () => { + expect(source).toMatch(/from\s*["']@\/lib\/jobs\/strain-score["']/); + expect(source).toMatch(/\bSTRAIN_SCORE_QUEUE\b/); + expect(source).toMatch(/\bSTRAIN_SCORE_CRON\b/); + expect(source).toMatch(/\brunStrainScore\b/); + }); + + it("registers the strain-score queue in allQueues", () => { + expect(allQueuesBlock()).toMatch(/\bSTRAIN_SCORE_QUEUE\b/); + }); + + it("schedules the strain-score cron", () => { + expect(source).toMatch(/\[STRAIN_SCORE_QUEUE,\s*STRAIN_SCORE_CRON\]/); + }); + + it("registers a boss.work handler that runs the strain-score pass", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,200}STRAIN_SCORE_QUEUE[\s\S]{0,400}runStrainScore/, + ); + }); +}); + +describe("reminder-worker — dense intra-day retention wiring", () => { + it("imports the queue symbols from the dense-intraday-retention module", () => { + expect(source).toMatch( + /from\s*["']@\/lib\/jobs\/dense-intraday-retention["']/, + ); + expect(source).toMatch(/\bDENSE_INTRADAY_RETENTION_QUEUE\b/); + expect(source).toMatch(/\brunDenseIntradayRetentionForUser\b/); + expect(source).toMatch(/\benqueueBootTimeDenseIntradayRetention\b/); + }); + + it("registers the dense-intraday-retention queue in allQueues", () => { + expect(allQueuesBlock()).toMatch(/\bDENSE_INTRADAY_RETENTION_QUEUE\b/); + }); + + it("registers a boss.work handler against the dense-intraday-retention queue", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,250}DENSE_INTRADAY_RETENTION_QUEUE[\s\S]{0,500}runDenseIntradayRetentionForUser/, + ); + }); + + it("fires the boot-discovery enqueue helper", () => { + expect(source).toMatch(/await enqueueBootTimeDenseIntradayRetention\(\)/); + }); + + it("folds the nightly dense-retention walk onto the drain-cumulative tick", () => { + // The steady-state nightly walk runs the global (no-userId) pass on the + // same tick as the cumulative + mean drains. + expect(source).toMatch(/runDenseIntradayRetention\(\s*getWorkerPrisma\(\)/); + }); +}); diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 4bf3ad5ad..86bf7ebb1 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -73,6 +73,24 @@ import { RECOVERY_SCORE_CRON, runRecoveryScore, } from "@/lib/jobs/recovery-score"; +import { + STRESS_SCORE_QUEUE, + STRESS_SCORE_CRON, + runStressScore, +} from "@/lib/jobs/stress-score"; +import { + STRAIN_SCORE_QUEUE, + STRAIN_SCORE_CRON, + runStrainScore, +} from "@/lib/jobs/strain-score"; +import { + DENSE_INTRADAY_RETENTION_QUEUE, + DENSE_INTRADAY_RETENTION_CONCURRENCY, + runDenseIntradayRetentionForUser, + enqueueBootTimeDenseIntradayRetention, + type DenseIntradayRetentionPayload, +} from "@/lib/jobs/dense-intraday-retention"; +import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; import { INSIGHT_STATUS_GENERATE_QUEUE, INSIGHT_STATUS_GENERATE_CONCURRENCY, @@ -2138,6 +2156,20 @@ export async function startReminderWorker() { // per eligible user. The queue MUST be registered here or pg-boss never // provisions it and the 04:45 schedule silently never fires. RECOVERY_SCORE_QUEUE, + // v1.10.0 — computed scores (WX-E). Nightly Stress-score (HRV-derived + // proxy) compute + store. Same createQueue contract as the recovery + // score; without this entry the 04:50 schedule silently never fires. + STRESS_SCORE_QUEUE, + // v1.10.0 — computed scores (WX-E). Nightly Strain-score (Banister TRIMP + // cardio-load) compute + store. Same createQueue contract; without this + // entry the 04:55 schedule silently never fires. + STRAIN_SCORE_QUEUE, + // v1.10.0 — computed scores (WX-E). Dense intra-day retention drain of + // daytime HRV / HR samples (per-user backfill queue, boot-discovery + // driven like mean-consolidation; the steady-state nightly walk folds + // onto the drain-cumulative tick). The queue MUST be registered here or + // the boot enqueue silently never drains. + DENSE_INTRADAY_RETENTION_QUEUE, ]; for (const q of allQueues) { @@ -2232,6 +2264,14 @@ export async function startReminderWorker() { // Recovery-score compute + store, after the rollup-feeding consolidation // + drain so the signals it reads are already folded. [RECOVERY_SCORE_QUEUE, RECOVERY_SCORE_CRON], + // v1.10.0 — computed scores (WX-E). Nightly 04:50 Europe/Berlin + // Stress-score compute + store, after the dense intra-day retention + // drain so the HRV inputs it reads are settled. + [STRESS_SCORE_QUEUE, STRESS_SCORE_CRON], + // v1.10.0 — computed scores (WX-E). Nightly 04:55 Europe/Berlin + // Strain-score compute + store, after the recovery + stress passes so + // the nightly score writes stay ordered. + [STRAIN_SCORE_QUEUE, STRAIN_SCORE_CRON], ]; for (const [name, cron] of schedules) { @@ -2410,6 +2450,48 @@ export async function startReminderWorker() { } }, ); + // v1.10.0 — computed scores (WX-E). Nightly Stress-score (HRV-derived + // proxy) compute + store. Single-flight so two ticks never double-walk + // the cohort. The runner iterates every eligible user and upserts one + // `COMPUTED STRESS_SCORE` row per scored day (idempotent — a re-fire + // overwrites in place). + await boss.work( + STRESS_SCORE_QUEUE, + { localConcurrency: 1 }, + async () => { + try { + const summary = await runStressScore(getWorkerPrisma()); + workerLog( + "info", + `[stress-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, + ); + } catch (err) { + recordError(); + workerLog("error", "[stress-score] pass failed", err); + throw err; + } + }, + ); + // v1.10.0 — computed scores (WX-E). Nightly Strain-score (Banister TRIMP + // cardio-load) compute + store. Single-flight; upserts one `COMPUTED + // STRAIN_SCORE` row per scored day (idempotent). + await boss.work( + STRAIN_SCORE_QUEUE, + { localConcurrency: 1 }, + async () => { + try { + const summary = await runStrainScore(getWorkerPrisma()); + workerLog( + "info", + `[strain-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`, + ); + } catch (err) { + recordError(); + workerLog("error", "[strain-score] pass failed", err); + throw err; + } + }, + ); // v1.8.3 — on-demand per-metric status generation enqueued by the // read-only status route on a cold card. Low concurrency so a first // visit that cold-misses several cards can't saturate the Prisma pool @@ -2527,6 +2609,38 @@ export async function startReminderWorker() { }, ); + // v1.10.0 WX-E — dense intra-day retention per-user backfill worker. The + // boot enqueue helper below sends one job per user holding live per-sample + // dense-tier (HRV / HR) rows older than the retention window; this handler + // folds those out-of-window samples to a daily mean and soft-deletes the + // originals, keeping the in-window intra-day shape intact for the Stress + // engine. Serial concurrency so the backfill never crowds the request pool. + await boss.work( + DENSE_INTRADAY_RETENTION_QUEUE, + { localConcurrency: DENSE_INTRADAY_RETENTION_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + try { + const { daysConsolidated, perSampleRowsSoftDeleted } = + await runDenseIntradayRetentionForUser(userId); + workerLog( + "info", + `[dense-intraday-retention] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted}`, + ); + } catch (err) { + recordError(); + workerLog( + "error", + `[dense-intraday-retention] user=${userId} failed`, + err, + ); + throw err; + } + } + }, + ); + // v1.8.2 — duplicate dose-slot cleanup worker. The boot enqueue helper // below sends one job per user holding two live intake rows that snap // to the same canonical slot; this handler keeps the winner, soft- @@ -2693,6 +2807,38 @@ export async function startReminderWorker() { recordError(); workerLog("error", "[mean-consolidation] nightly run failed", err); } + + // v1.10.0 WX-E — fold the dense intra-day retention drain onto the + // same nightly tick. Unlike the daily-mean drain, this scopes to the + // dense-tier types (HEART_RATE_VARIABILITY, PULSE) and keeps the last + // DENSE_INTRADAY_RETENTION_DAYS of raw per-sample rows so the Stress + // engine still sees the intra-day SDNN shape; only out-of-window + // samples fold to a daily mean. These two types are NEVER in the + // destructive HIGH_FREQUENCY_MEAN_TYPES allowlist — the drain + // exemption the intra-day shape depends on. The global (no `userId`) + // signature drains every user. Boot discovery (below) back-fills + // accounts that accumulated out-of-window raw rows before this + // shipped. + try { + const denseSummary = await runDenseIntradayRetention( + getWorkerPrisma(), + { + dryRun: false, + log: (line) => workerLog("info", line), + }, + ); + workerLog( + "info", + `[dense-intraday-retention] triggeredAt=${job.data.triggeredAt} usersScanned=${denseSummary.totals.usersScanned} daysConsolidated=${denseSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${denseSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${denseSummary.totals.dailyRowsUpserted}`, + ); + } catch (err) { + recordError(); + workerLog( + "error", + "[dense-intraday-retention] nightly run failed", + err, + ); + } } }, ); @@ -2789,6 +2935,34 @@ export async function startReminderWorker() { ); } + // v1.10.0 WX-E — fire-and-forget boot discovery for the dense intra-day + // retention drain. Finds every user holding live per-sample dense-tier + // (HRV / HR) rows OLDER than the retention window and enqueues one job per + // account. Idempotent across reboots: folded rows are soft-deleted, so the + // discovery predicate drops them. Errors are returned through the helper's + // result value — the worker boot never fails on a miss. + try { + const { enqueued, skipped, error } = + await enqueueBootTimeDenseIntradayRetention(); + if (error) { + workerLog( + "error", + `[dense-intraday-retention] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[dense-intraday-retention] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[dense-intraday-retention] boot discovery threw an unexpected error", + err, + ); + } + // v1.8.2 — fire-and-forget boot discovery for the duplicate dose-slot // cleanup. Finds every user holding two live intake rows within the // drift window on the same medication and enqueues one dedup job per