feat(insights): register the stress, strain, and dense-retention jobs

Wire the two nightly score passes and the dense intra-day retention drain
into the worker. The stress (04:50) and strain (04:55) crons each get a
queue, a schedule, and a single-flight handler; the dense-retention drain
gets a per-user backfill queue with boot discovery (like mean-consolidation)
and folds its steady-state nightly walk onto the existing drain-cumulative
tick, after the daily-mean fold.

A queue not registered in allQueues silently never drains, so a source-grep
guard pins every registration site for all three jobs.
This commit is contained in:
Marc-André Bombeck committed 2026-06-02 22:49:43 +02:00
1 parent 6935feda15
commit c5ba38cebb
2 files changed
+275

No files matched your search

@@ -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\(\)/);
});
});
+174
View File
@@ -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<DenseIntradayRetentionPayload>(
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