mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-12 22:26:13 +00:00
Merge branch 'feat/v1462-scale' into release/v1.46.2
# Conflicts: # src/__tests__/baselines/version-stamp-comments.json
This commit is contained in:
Marc-André Bombeck committed
commit
ad5d154561
57 files changed
+2736
-425
No files matched your search
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
|
||||
"count": 84,
|
||||
"total": 94928,
|
||||
"total": 94686,
|
||||
"entries": {
|
||||
"app/medications/page-client.tsx": 839,
|
||||
"components/charts/mood-chart.tsx": 1166,
|
||||
@@ -71,14 +71,14 @@
|
||||
"lib/managed-profiles/handover.ts": 986,
|
||||
"lib/mcp/record-search.ts": 1467,
|
||||
"lib/mcp/rich-reads.ts": 1467,
|
||||
"lib/measurements/dense-intraday-retention.ts": 1057,
|
||||
"lib/measurements/dense-intraday-retention.ts": 1010,
|
||||
"lib/measurements/import-apple-health-export.ts": 1722,
|
||||
"lib/medications/scheduling/cadence.ts": 976,
|
||||
"lib/medications/scheduling/recurrence.ts": 1100,
|
||||
"lib/medications/scheduling/slot-upsert.ts": 912,
|
||||
"lib/notifications/senders/apns.ts": 1100,
|
||||
"lib/oura/sync.ts": 900,
|
||||
"lib/rollups/measurement-rollups.ts": 1496,
|
||||
"lib/rollups/measurement-rollups.ts": 1302,
|
||||
"lib/search/search-records.ts": 996,
|
||||
"lib/sharing/grants.ts": 806,
|
||||
"lib/support/report.ts": 903,
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.",
|
||||
"count": 2896,
|
||||
"total": 9235,
|
||||
"total": 9229,
|
||||
"entries": {
|
||||
"__tests__/account-payload-consumer-guard.test.ts": 2,
|
||||
"__tests__/acting-account-boundary-guard.test.ts": 7,
|
||||
@@ -175,9 +175,9 @@
|
||||
"app/api/auth/refresh/route.ts": 2,
|
||||
"app/api/auth/register/__tests__/route.test.ts": 5,
|
||||
"app/api/auth/register/route.ts": 10,
|
||||
"app/api/biomarkers/__tests__/error-contract.test.ts": 1,
|
||||
"app/api/biomarkers/[id]/__tests__/route.test.ts": 2,
|
||||
"app/api/biomarkers/[id]/route.ts": 6,
|
||||
"app/api/biomarkers/__tests__/error-contract.test.ts": 1,
|
||||
"app/api/biomarkers/route.ts": 4,
|
||||
"app/api/c/[token]/unlock/__tests__/route.test.ts": 1,
|
||||
"app/api/c/[token]/unlock/route.ts": 1,
|
||||
@@ -187,12 +187,12 @@
|
||||
"app/api/coach/about-me/questions/__tests__/route-module-gate.test.ts": 1,
|
||||
"app/api/coach/about-me/questions/route.ts": 3,
|
||||
"app/api/coach/about-me/route.ts": 4,
|
||||
"app/api/coach/plans/[id]/route.ts": 2,
|
||||
"app/api/coach/plans/__tests__/route.test.ts": 1,
|
||||
"app/api/coach/plans/[id]/route.ts": 2,
|
||||
"app/api/coach/plans/route.ts": 1,
|
||||
"app/api/coach/reminder-suggestions/route.ts": 1,
|
||||
"app/api/coach/reminders/[id]/route.ts": 1,
|
||||
"app/api/coach/reminders/__tests__/route.test.ts": 1,
|
||||
"app/api/coach/reminders/[id]/route.ts": 1,
|
||||
"app/api/coach/reminders/route.ts": 1,
|
||||
"app/api/coach/suggested-actions/__tests__/route.test.ts": 1,
|
||||
"app/api/coach/suggested-actions/route.ts": 1,
|
||||
@@ -200,9 +200,9 @@
|
||||
"app/api/consent/ai/latest/__tests__/route.test.ts": 3,
|
||||
"app/api/consent/ai/latest/route.ts": 3,
|
||||
"app/api/consent/ai/route.ts": 3,
|
||||
"app/api/custom-metrics/[id]/entries/[entryId]/route.ts": 2,
|
||||
"app/api/custom-metrics/[id]/entries/__tests__/note-encryption.test.ts": 1,
|
||||
"app/api/custom-metrics/[id]/entries/__tests__/soft-delete-restore.test.ts": 1,
|
||||
"app/api/custom-metrics/[id]/entries/[entryId]/route.ts": 2,
|
||||
"app/api/custom-metrics/[id]/entries/route.ts": 4,
|
||||
"app/api/custom-metrics/[id]/route.ts": 1,
|
||||
"app/api/custom-metrics/route.ts": 1,
|
||||
@@ -212,8 +212,8 @@
|
||||
"app/api/cycle/day-logs/[id]/route.ts": 4,
|
||||
"app/api/cycle/day-logs/route.ts": 1,
|
||||
"app/api/cycle/period/route.ts": 4,
|
||||
"app/api/cycle/symptoms/custom/[key]/route.ts": 1,
|
||||
"app/api/cycle/symptoms/custom/__tests__/custom-crud.test.ts": 1,
|
||||
"app/api/cycle/symptoms/custom/[key]/route.ts": 1,
|
||||
"app/api/cycle/symptoms/custom/route.ts": 3,
|
||||
"app/api/dashboard/chart-overlay-prefs/__tests__/route.test.ts": 1,
|
||||
"app/api/dashboard/chart-overlay-prefs/route.ts": 2,
|
||||
@@ -223,10 +223,11 @@
|
||||
"app/api/dashboard/summary/route.ts": 5,
|
||||
"app/api/dashboard/widgets/__tests__/route.test.ts": 4,
|
||||
"app/api/dashboard/widgets/route.ts": 14,
|
||||
"app/api/devices/__tests__/route.test.ts": 1,
|
||||
"app/api/devices/[id]/__tests__/route.test.ts": 1,
|
||||
"app/api/devices/[id]/route.ts": 1,
|
||||
"app/api/devices/__tests__/route.test.ts": 1,
|
||||
"app/api/devices/route.ts": 4,
|
||||
"app/api/documents/inbound/__tests__/route.test.ts": 1,
|
||||
"app/api/documents/inbound/[id]/__tests__/route.test.ts": 1,
|
||||
"app/api/documents/inbound/[id]/chat/__tests__/route.test.ts": 2,
|
||||
"app/api/documents/inbound/[id]/chat/route.ts": 1,
|
||||
@@ -241,7 +242,6 @@
|
||||
"app/api/documents/inbound/[id]/suggest/route.ts": 2,
|
||||
"app/api/documents/inbound/[id]/summary/route.ts": 2,
|
||||
"app/api/documents/inbound/[id]/thumbnail/route.ts": 1,
|
||||
"app/api/documents/inbound/__tests__/route.test.ts": 1,
|
||||
"app/api/documents/inbound/reindex/route.ts": 1,
|
||||
"app/api/documents/inbound/route.ts": 2,
|
||||
"app/api/documents/inbound/usage/route.ts": 1,
|
||||
@@ -260,10 +260,10 @@
|
||||
"app/api/export/mood/route.ts": 2,
|
||||
"app/api/family-history/[id]/route.ts": 2,
|
||||
"app/api/family-history/route.ts": 1,
|
||||
"app/api/fhir/Patient/$everything/route.ts": 1,
|
||||
"app/api/fhir/__tests__/observation-series.test.ts": 1,
|
||||
"app/api/fhir/__tests__/route.test.ts": 1,
|
||||
"app/api/fhir/metadata/route.ts": 1,
|
||||
"app/api/fhir/Patient/$everything/route.ts": 1,
|
||||
"app/api/fitbit/callback/route.ts": 1,
|
||||
"app/api/fitbit/connect/route.ts": 1,
|
||||
"app/api/fitbit/credentials/route.ts": 1,
|
||||
@@ -292,11 +292,6 @@
|
||||
"app/api/insights/breathing-screening/route.ts": 2,
|
||||
"app/api/insights/cards/__tests__/route.test.ts": 2,
|
||||
"app/api/insights/cards/route.ts": 9,
|
||||
"app/api/insights/chat/[id]/attachments/[documentId]/route.ts": 1,
|
||||
"app/api/insights/chat/[id]/attachments/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/chat/[id]/attachments/route.ts": 1,
|
||||
"app/api/insights/chat/[id]/messages/[messageId]/trail/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/chat/[id]/route.ts": 1,
|
||||
"app/api/insights/chat/__tests__/dialog-harness.ts": 1,
|
||||
"app/api/insights/chat/__tests__/golden-transcripts.test.ts": 1,
|
||||
"app/api/insights/chat/__tests__/route-activity.test.ts": 1,
|
||||
@@ -308,14 +303,19 @@
|
||||
"app/api/insights/chat/__tests__/route-snapshot-once.test.ts": 2,
|
||||
"app/api/insights/chat/__tests__/route-tool-mode.test.ts": 4,
|
||||
"app/api/insights/chat/__tests__/route-workout-scope.test.ts": 1,
|
||||
"app/api/insights/chat/[id]/attachments/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/chat/[id]/attachments/[documentId]/route.ts": 1,
|
||||
"app/api/insights/chat/[id]/attachments/route.ts": 1,
|
||||
"app/api/insights/chat/[id]/messages/[messageId]/trail/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/chat/[id]/route.ts": 1,
|
||||
"app/api/insights/chat/fenced/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/chat/fenced/route.ts": 1,
|
||||
"app/api/insights/chat/messages/[id]/feedback/route.ts": 3,
|
||||
"app/api/insights/chat/route.ts": 9,
|
||||
"app/api/insights/coach-read/route.ts": 2,
|
||||
"app/api/insights/coach/facts/[id]/route.ts": 2,
|
||||
"app/api/insights/coach/facts/__tests__/route-memory.test.ts": 1,
|
||||
"app/api/insights/coach/facts/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/coach/facts/[id]/route.ts": 2,
|
||||
"app/api/insights/coach/facts/route.ts": 1,
|
||||
"app/api/insights/coach/nudge-status/route.ts": 1,
|
||||
"app/api/insights/coach/seen/route.ts": 1,
|
||||
@@ -326,9 +326,9 @@
|
||||
"app/api/insights/derived/batch/__tests__/route.test.ts": 3,
|
||||
"app/api/insights/derived/batch/route.ts": 6,
|
||||
"app/api/insights/derived/route.ts": 5,
|
||||
"app/api/insights/ecg/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/ecg/[id]/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/ecg/[id]/route.ts": 2,
|
||||
"app/api/insights/ecg/__tests__/route.test.ts": 1,
|
||||
"app/api/insights/ecg/route.ts": 1,
|
||||
"app/api/insights/feedback/route.ts": 2,
|
||||
"app/api/insights/generate/__tests__/route.test.ts": 12,
|
||||
@@ -367,8 +367,8 @@
|
||||
"app/api/internal/deploy-webhook/route.ts": 1,
|
||||
"app/api/internal/web-vitals/__tests__/route.test.ts": 1,
|
||||
"app/api/internal/web-vitals/route.ts": 1,
|
||||
"app/api/labs/[id]/route.ts": 7,
|
||||
"app/api/labs/__tests__/error-contract.test.ts": 1,
|
||||
"app/api/labs/[id]/route.ts": 7,
|
||||
"app/api/labs/ocr/capability/route.ts": 1,
|
||||
"app/api/labs/ocr/commit/__tests__/route.test.ts": 2,
|
||||
"app/api/labs/ocr/commit/route.ts": 4,
|
||||
@@ -388,13 +388,13 @@
|
||||
"app/api/measurement-reminders/[id]/skip/route.ts": 1,
|
||||
"app/api/measurement-reminders/[id]/snooze/route.ts": 1,
|
||||
"app/api/measurement-reminders/route.ts": 3,
|
||||
"app/api/measurements/[id]/__tests__/route.test.ts": 2,
|
||||
"app/api/measurements/[id]/route.ts": 16,
|
||||
"app/api/measurements/__tests__/arrival-wiring.test.ts": 1,
|
||||
"app/api/measurements/__tests__/group-by-day.test.ts": 8,
|
||||
"app/api/measurements/__tests__/put-duplicate-timestamp.test.ts": 1,
|
||||
"app/api/measurements/__tests__/range-aggregation-route.test.ts": 3,
|
||||
"app/api/measurements/__tests__/zod-multi-issue.test.ts": 4,
|
||||
"app/api/measurements/[id]/__tests__/route.test.ts": 2,
|
||||
"app/api/measurements/[id]/route.ts": 16,
|
||||
"app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts": 1,
|
||||
"app/api/measurements/batch/__tests__/duplicate-prefilter.test.ts": 2,
|
||||
"app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts": 1,
|
||||
@@ -411,6 +411,7 @@
|
||||
"app/api/measurements/series-batch/route.ts": 1,
|
||||
"app/api/measurements/series/__tests__/route.test.ts": 5,
|
||||
"app/api/measurements/series/route.ts": 6,
|
||||
"app/api/medications/__tests__/route.test.ts": 10,
|
||||
"app/api/medications/[id]/__tests__/route.test.ts": 8,
|
||||
"app/api/medications/[id]/api-endpoint/route.ts": 2,
|
||||
"app/api/medications/[id]/cadence/route.ts": 10,
|
||||
@@ -422,10 +423,10 @@
|
||||
"app/api/medications/[id]/dose-history/route.ts": 14,
|
||||
"app/api/medications/[id]/glp1/__tests__/route.test.ts": 3,
|
||||
"app/api/medications/[id]/glp1/route.ts": 8,
|
||||
"app/api/medications/[id]/intake/[eventId]/__tests__/route.test.ts": 9,
|
||||
"app/api/medications/[id]/intake/[eventId]/route.ts": 4,
|
||||
"app/api/medications/[id]/intake/__tests__/route.test.ts": 11,
|
||||
"app/api/medications/[id]/intake/__tests__/source-provenance.test.ts": 1,
|
||||
"app/api/medications/[id]/intake/[eventId]/__tests__/route.test.ts": 9,
|
||||
"app/api/medications/[id]/intake/[eventId]/route.ts": 4,
|
||||
"app/api/medications/[id]/intake/bulk-delete/route.ts": 7,
|
||||
"app/api/medications/[id]/intake/import/__tests__/route.test.ts": 2,
|
||||
"app/api/medications/[id]/intake/import/route.ts": 6,
|
||||
@@ -433,12 +434,11 @@
|
||||
"app/api/medications/[id]/inventory/[itemId]/route.ts": 8,
|
||||
"app/api/medications/[id]/inventory/route.ts": 4,
|
||||
"app/api/medications/[id]/phase-config/route.ts": 4,
|
||||
"app/api/medications/[id]/schedule-revisions/[revisionId]/route.ts": 9,
|
||||
"app/api/medications/[id]/schedule-revisions/__tests__/route.test.ts": 1,
|
||||
"app/api/medications/[id]/schedule-revisions/[revisionId]/route.ts": 9,
|
||||
"app/api/medications/[id]/schedule-revisions/route.ts": 3,
|
||||
"app/api/medications/[id]/side-effects/[logId]/route.ts": 2,
|
||||
"app/api/medications/[id]/side-effects/route.ts": 5,
|
||||
"app/api/medications/__tests__/route.test.ts": 10,
|
||||
"app/api/medications/categories/[key]/route.ts": 1,
|
||||
"app/api/medications/categories/route.ts": 1,
|
||||
"app/api/medications/compliance/__tests__/route.test.ts": 2,
|
||||
@@ -455,9 +455,9 @@
|
||||
"app/api/medications/route.ts": 2,
|
||||
"app/api/mental-health/assessments/__tests__/route.test.ts": 2,
|
||||
"app/api/mental-health/assessments/route.ts": 6,
|
||||
"app/api/mood-entries/__tests__/route.test.ts": 7,
|
||||
"app/api/mood-entries/[id]/__tests__/route.test.ts": 4,
|
||||
"app/api/mood-entries/[id]/route.ts": 22,
|
||||
"app/api/mood-entries/__tests__/route.test.ts": 7,
|
||||
"app/api/mood-entries/bulk-delete/__tests__/route.test.ts": 1,
|
||||
"app/api/mood-entries/bulk-delete/route.ts": 4,
|
||||
"app/api/mood-entries/bulk/__tests__/route.test.ts": 5,
|
||||
@@ -547,12 +547,12 @@
|
||||
"app/api/withings/connect/__tests__/route.test.ts": 1,
|
||||
"app/api/withings/connect/route.ts": 3,
|
||||
"app/api/withings/status/route.ts": 1,
|
||||
"app/api/withings/webhook/[token]/__tests__/route.test.ts": 1,
|
||||
"app/api/withings/webhook/__tests__/route.test.ts": 1,
|
||||
"app/api/withings/webhook/[token]/__tests__/route.test.ts": 1,
|
||||
"app/api/withings/webhook/route.ts": 2,
|
||||
"app/api/workouts/[id]/route.ts": 2,
|
||||
"app/api/workouts/__tests__/batch-create.test.ts": 3,
|
||||
"app/api/workouts/__tests__/canonical-dedup.test.ts": 3,
|
||||
"app/api/workouts/[id]/route.ts": 2,
|
||||
"app/api/workouts/batch/route.ts": 3,
|
||||
"app/api/workouts/route.ts": 2,
|
||||
"app/auth/claim/page.tsx": 1,
|
||||
@@ -654,20 +654,20 @@
|
||||
"app/loading.tsx": 2,
|
||||
"app/mcp/__tests__/route.test.ts": 5,
|
||||
"app/measurements/page.tsx": 5,
|
||||
"app/medications/[id]/history/page.tsx": 1,
|
||||
"app/medications/[id]/page.tsx": 2,
|
||||
"app/medications/__tests__/category-filter.test.tsx": 2,
|
||||
"app/medications/__tests__/detail-page-history-surface.test.ts": 1,
|
||||
"app/medications/__tests__/list-view-toggle.test.tsx": 5,
|
||||
"app/medications/__tests__/page-prefetch.test.tsx": 1,
|
||||
"app/medications/__tests__/take-all-due-affordance.test.tsx": 1,
|
||||
"app/medications/[id]/history/page.tsx": 1,
|
||||
"app/medications/[id]/page.tsx": 2,
|
||||
"app/medications/new/page.tsx": 1,
|
||||
"app/medications/page-client.tsx": 50,
|
||||
"app/mental-wellbeing/page.tsx": 1,
|
||||
"app/mood/page-client.tsx": 2,
|
||||
"app/not-found.tsx": 2,
|
||||
"app/onboarding/[step]/page.tsx": 1,
|
||||
"app/onboarding/__tests__/page.test.tsx": 1,
|
||||
"app/onboarding/[step]/page.tsx": 1,
|
||||
"app/onboarding/handover/page.tsx": 1,
|
||||
"app/onboarding/page.tsx": 2,
|
||||
"app/privacy/__tests__/page.test.tsx": 1,
|
||||
@@ -2747,7 +2747,7 @@
|
||||
"lib/rollups/measurement-coverage.ts": 1,
|
||||
"lib/rollups/measurement-read-wmy.ts": 4,
|
||||
"lib/rollups/measurement-read.ts": 10,
|
||||
"lib/rollups/measurement-rollups.ts": 18,
|
||||
"lib/rollups/measurement-rollups.ts": 12,
|
||||
"lib/rollups/mood-rollups.ts": 4,
|
||||
"lib/rollups/tiered-context.ts": 6,
|
||||
"lib/score-config/rows.ts": 1,
|
||||
|
||||
@@ -163,6 +163,11 @@ const ALLOWED: Readonly<Record<string, { count: number; reason: string }>> = {
|
||||
count: 1,
|
||||
reason: "No /proc/diskstats outside Linux; see captureHostMetric.",
|
||||
},
|
||||
"lib/jobs/system-job-alert.ts::readTerminalRuns": {
|
||||
count: 1,
|
||||
reason:
|
||||
"No pgboss schema on a web-only deployment; null means no queue to ask, and the evaluation then judges nothing.",
|
||||
},
|
||||
"lib/jobs/job-failures.ts::readActiveJobsStartedBefore": {
|
||||
count: 1,
|
||||
reason:
|
||||
|
||||
@@ -15,8 +15,8 @@
|
||||
* { userId?: string }
|
||||
*
|
||||
* - `userId` — single user mode. Synchronously calls
|
||||
* `recomputeUserRollups(userId)` and returns the row count when
|
||||
* the fold completes. The call is awaited inside the request, so
|
||||
* `foldUserRollupHistory(userId)` (the full fold, a calendar year per
|
||||
* statement) and returns the row count when the fold completes. The call is awaited inside the request, so
|
||||
* the operator sees the duration in the response.
|
||||
* - `userId` omitted — calls `enqueueBootTimeRollupBackfill()` to
|
||||
* re-run the worker-boot discovery (`{ enqueued, skipped }` shape).
|
||||
@@ -37,8 +37,8 @@ import {
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import {
|
||||
enqueueBootTimeRollupBackfill,
|
||||
recomputeUserRollups,
|
||||
} from "@/lib/rollups/measurement-rollups";
|
||||
foldUserRollupHistory,
|
||||
} from "@/lib/rollups/full-backfill";
|
||||
|
||||
interface RecomputeBody {
|
||||
userId?: string;
|
||||
@@ -68,9 +68,9 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
details: { targetUserId: userId, mode: "single" },
|
||||
});
|
||||
|
||||
let result: Awaited<ReturnType<typeof recomputeUserRollups>>;
|
||||
let result: Awaited<ReturnType<typeof foldUserRollupHistory>>;
|
||||
try {
|
||||
result = await recomputeUserRollups(userId);
|
||||
result = await foldUserRollupHistory(userId);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
return apiError(message, 500);
|
||||
|
||||
@@ -316,6 +316,8 @@ export async function computeRangeDelta(
|
||||
userId,
|
||||
type,
|
||||
windowDays * 2,
|
||||
undefined,
|
||||
new Date(now),
|
||||
);
|
||||
if (!resolved) {
|
||||
const empty: WindowAggregate = {
|
||||
|
||||
@@ -358,6 +358,8 @@ export const INSTANCE_SCOPED: Readonly<Record<string, string>> = {
|
||||
"the monthly restore drill's last verdict for the instance's off-host bucket; says nothing about any account",
|
||||
RecoveryReadinessNotice:
|
||||
"per backup signal, what the admins were last told; a fact about the instance's backups, not about any account",
|
||||
SystemJobAlertNotice:
|
||||
"per background queue, what the admins were last told about its failures; a fact about the instance's jobs, not about any account",
|
||||
RecoveryKitConfirmation:
|
||||
"the instance's recovery kit confirmation (key ids and fingerprints); the confirming admin is a plain id, not a relation",
|
||||
DatabaseBackupRun:
|
||||
|
||||
@@ -89,7 +89,7 @@ Scope: `googlehealth.health_metrics_and_measurements.readonly`.
|
||||
| `daily-heart-rate-variability` | `dailyHeartRateVariability.averageHeartRateVariabilityMilliseconds` | `HRV_RMSSD` | ms | `hrv` | daily | Google documents the daily average as RMSSD, so it lands in `HRV_RMSSD`, apart from the SDNN `HEART_RATE_VARIABILITY`. |
|
||||
| `daily-resting-heart-rate` | `dailyRestingHeartRate.beatsPerMinute` (int64 string) | `RESTING_HEART_RATE` | bpm | `rhr` | daily | |
|
||||
| `daily-respiratory-rate` | `dailyRespiratoryRate.breathsPerMinute` (number) | `RESPIRATORY_RATE` | breaths/min | `resp_rate` | daily | `respiratory-rate` does NOT exist in the catalogue. Schema is `{date, breathsPerMinute}` — the earlier `dailyRespiratoryRateBpm` leaf never existed. |
|
||||
| `heart-rate` | `heartRate.beatsPerMinute` (int64 string) | `PULSE` | bpm | `hr` | sample | Intraday spot HR; per-minute volume — consider the 14-day-max rollup if PULSE should stay coarse. |
|
||||
| `heart-rate` | `heartRate.beatsPerMinute` (int64 string) | `PULSE` | bpm | `hr` | sample | Intraday spot HR, one row per sample; older than 90 days it folds to hourly means (`dense-intraday-sources.ts`). |
|
||||
| `blood-glucose` | `bloodGlucose.bloodGlucoseMilligramsPerDeciliter` (number) | `BLOOD_GLUCOSE` | mg/dL | `glucose` | sample | Already in the canonical mg/dL storage unit. Meal/timing/specimen enums not stored. |
|
||||
| `core-body-temperature` | `coreBodyTemperature.temperatureCelsius` (number) | `BODY_TEMPERATURE` | celsius | `core_temp` | sample | Core slot — distinct from SKIN_TEMPERATURE / WRIST_TEMPERATURE. |
|
||||
| `daily-sleep-temperature-derivations` | `dailySleepTemperatureDerivations.nightlyTemperatureCelsius` (number) | `WRIST_TEMPERATURE` | celsius | `wrist_temp` | daily | ABSOLUTE nightly skin temperature (mean over the night's samples) — the schema refutes the earlier "deviation-only" premise. Baseline/stddev derivations not stored. |
|
||||
|
||||
@@ -36,6 +36,7 @@ import {
|
||||
type MeasurementReconciliationVerdict,
|
||||
} from "@/lib/measurements/reconcile-external-measurement";
|
||||
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
|
||||
import { withoutFoldedSamples } from "@/lib/measurements/folded-window";
|
||||
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
|
||||
import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status";
|
||||
import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups";
|
||||
@@ -1128,7 +1129,7 @@ export async function upsertGoogleHealthMeasurements(
|
||||
// before they can become rows. A refused reading is dropped and tallied on
|
||||
// the ambient event, never clamped and never written for a later reader to
|
||||
// discover.
|
||||
const { kept: readings, dropped: naturalKeyTwins } = collapseNaturalKeyTwins(
|
||||
const { kept: collapsed, dropped: naturalKeyTwins } = collapseNaturalKeyTwins(
|
||||
dropImplausibleMeasurements("googleHealth", incoming, (r) => ({
|
||||
type: r.type,
|
||||
value: r.value,
|
||||
@@ -1137,14 +1138,18 @@ export async function upsertGoogleHealthMeasurements(
|
||||
if (naturalKeyTwins > 0) {
|
||||
annotate({ meta: { "googleHealth.naturalKeyTwins": naturalKeyTwins } });
|
||||
}
|
||||
// A sample of an hour the dense fold already took is not written again.
|
||||
const { kept: readings } = await withoutFoldedSamples(
|
||||
prisma,
|
||||
userId,
|
||||
"GOOGLE_HEALTH",
|
||||
collapsed,
|
||||
);
|
||||
if (readings.length === 0) return { imported: 0, touched: [], inserted: [] };
|
||||
|
||||
// Probe EVERY existing row (live AND tombstoned) for the batch's externalIds
|
||||
// in a single query. The full unique index guarantees at most one row per
|
||||
// `(type, externalId)` key, and a tombstoned match must take the UPDATE
|
||||
// branch (which clears `deletedAt`) — treating it as absent would plan an
|
||||
// insert that `skipDuplicates` silently drops against the tombstone's key,
|
||||
// permanently wedging the key (see TOMBSTONES RESURRECT above).
|
||||
// in one query: a tombstoned match must take the UPDATE branch, or the
|
||||
// insert is silently dropped against its key (see TOMBSTONES RESURRECT).
|
||||
const externalIds = readings.map((r) => r.externalId);
|
||||
interface ProbedRow {
|
||||
id: string;
|
||||
@@ -1162,10 +1167,8 @@ export async function upsertGoogleHealthMeasurements(
|
||||
source: "GOOGLE_HEALTH",
|
||||
externalId: { in: externalIds },
|
||||
},
|
||||
// The payload fields ride along so the update branch can skip a no-op
|
||||
// overwrite: the 24 h overlap re-fetches every recent row hourly, and an
|
||||
// unconditional write would bump `syncVersion` (iOS delta churn) on rows
|
||||
// whose values did not change.
|
||||
// The payload rides along so the update branch can skip a no-op
|
||||
// overwrite (the hourly 24 h overlap would bump `syncVersion` for nothing).
|
||||
select: {
|
||||
id: true,
|
||||
type: true,
|
||||
@@ -1193,9 +1196,8 @@ export async function upsertGoogleHealthMeasurements(
|
||||
]),
|
||||
);
|
||||
} catch (err) {
|
||||
// A probe failure must not strand the whole batch; fall back to treating
|
||||
// every row as fresh (`skipDuplicates` absorbs the collision on the full
|
||||
// unique index — those rows simply retry on the next sync tick).
|
||||
// A probe failure must not strand the batch: every row is treated as
|
||||
// fresh, and `skipDuplicates` absorbs the collisions until the next tick.
|
||||
getEvent()?.addWarning(`google-health: row probe failed: ${err}`);
|
||||
}
|
||||
|
||||
@@ -1225,11 +1227,9 @@ export async function upsertGoogleHealthMeasurements(
|
||||
const key = `${type} ${r.externalId}`;
|
||||
const existingRow = rowByKey.get(key);
|
||||
if (existingRow) {
|
||||
// Skip the no-op overwrite: a LIVE row whose payload matches the
|
||||
// re-fetched reading exactly gains nothing from a write, and the
|
||||
// unconditional `syncVersion` bump churned the DB + every iOS delta pull
|
||||
// hourly for the whole 24 h overlap window. A TOMBSTONED row always
|
||||
// updates — the write is what resurrects it (`deletedAt: null`).
|
||||
// Skip the no-op overwrite of a LIVE row whose payload matches exactly
|
||||
// (it would churn `syncVersion` and every iOS delta pull). A TOMBSTONED
|
||||
// row always updates: the write is what resurrects it.
|
||||
const unchanged =
|
||||
existingRow.deletedAt === null &&
|
||||
existingRow.value === r.value &&
|
||||
|
||||
@@ -217,7 +217,13 @@ async function readWindowMeans(
|
||||
1,
|
||||
Math.ceil((to.getTime() - from.getTime()) / MS_PER_DAY),
|
||||
);
|
||||
const resolved = await readBestGranularityRollups(userId, type, windowDays);
|
||||
const resolved = await readBestGranularityRollups(
|
||||
userId,
|
||||
type,
|
||||
windowDays,
|
||||
undefined,
|
||||
now,
|
||||
);
|
||||
if (
|
||||
resolved &&
|
||||
resolved.granularity === "DAY" &&
|
||||
|
||||
@@ -274,7 +274,13 @@ export async function readDayMeanSeries(
|
||||
// band from WEEK/MONTH `sd`, and the center reads the DAY `mean`
|
||||
// which composes exactly. `readBestGranularityRollups` with a
|
||||
// window < 91 days resolves to DAY by construction.
|
||||
const resolved = await readBestGranularityRollups(userId, type, windowDays);
|
||||
const resolved = await readBestGranularityRollups(
|
||||
userId,
|
||||
type,
|
||||
windowDays,
|
||||
undefined,
|
||||
now,
|
||||
);
|
||||
if (
|
||||
resolved &&
|
||||
resolved.granularity === "DAY" &&
|
||||
|
||||
@@ -537,6 +537,7 @@ export async function buildGradedSeriesWithRollups(
|
||||
type,
|
||||
365,
|
||||
priority,
|
||||
now,
|
||||
);
|
||||
// Yearly slice: route a multi-year window through the tier.
|
||||
const yearlyRouted = await readBestGranularityRollups(
|
||||
@@ -544,6 +545,7 @@ export async function buildGradedSeriesWithRollups(
|
||||
type,
|
||||
1095,
|
||||
priority,
|
||||
now,
|
||||
);
|
||||
|
||||
const monthlyCovered =
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
/**
|
||||
* The system job alert's verdict on one queue's ledger (#1298). The notice
|
||||
* flow itself runs against a real pg-boss in
|
||||
* `tests/integration/system-job-alert.test.ts`.
|
||||
*/
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import {
|
||||
judgeQueueRuns,
|
||||
SYSTEM_JOB_FAILURE_STREAK,
|
||||
watchedQueues,
|
||||
type TerminalRun,
|
||||
} from "@/lib/jobs/system-job-alert";
|
||||
|
||||
const at = new Date("2026-10-29T10:00:00.000Z");
|
||||
|
||||
function run(overrides: Partial<TerminalRun> = {}): TerminalRun {
|
||||
return {
|
||||
state: "failed",
|
||||
retryCount: 0,
|
||||
allFailed: false,
|
||||
completedOn: at,
|
||||
error: "boom",
|
||||
expireSeconds: null,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe("judgeQueueRuns", () => {
|
||||
it("is inactive with no run left in the ledger", () => {
|
||||
expect(judgeQueueRuns([]).state).toBe("inactive");
|
||||
});
|
||||
|
||||
it("counts every attempt of a run, so one job that used up its retries is a streak", () => {
|
||||
const verdict = judgeQueueRuns([run({ retryCount: 3 })]);
|
||||
expect(verdict).toMatchObject({ state: "failing", failures: 4 });
|
||||
expect(verdict.lastError).toBe("boom");
|
||||
});
|
||||
|
||||
it("calls the streak at the threshold, not before", () => {
|
||||
const below = Array.from({ length: SYSTEM_JOB_FAILURE_STREAK - 1 }, () =>
|
||||
run(),
|
||||
);
|
||||
expect(judgeQueueRuns(below).state).toBe("pending");
|
||||
expect(judgeQueueRuns([...below, run()]).state).toBe("failing");
|
||||
});
|
||||
|
||||
it("ends the streak at the newest success", () => {
|
||||
const verdict = judgeQueueRuns([
|
||||
run({ error: "newest" }),
|
||||
run({ state: "completed", error: null }),
|
||||
run(),
|
||||
run(),
|
||||
run(),
|
||||
]);
|
||||
expect(verdict).toMatchObject({
|
||||
state: "ok",
|
||||
failures: 1,
|
||||
lastError: "newest",
|
||||
});
|
||||
});
|
||||
|
||||
it("counts a completed run that says every candidate failed", () => {
|
||||
const allFailed = run({ state: "completed", allFailed: true });
|
||||
expect(judgeQueueRuns([allFailed, allFailed, allFailed]).state).toBe(
|
||||
"failing",
|
||||
);
|
||||
});
|
||||
|
||||
it("explains a bare timeout the way the admin card does", () => {
|
||||
const verdict = judgeQueueRuns([
|
||||
run({ error: "job timed out", expireSeconds: 3600, retryCount: 2 }),
|
||||
]);
|
||||
expect(verdict.lastError).toMatch(
|
||||
/^job timed out: did not finish within its 1 h limit/,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("watchedQueues", () => {
|
||||
it("watches the system queues and none with an account-facing surface", () => {
|
||||
const queues = watchedQueues();
|
||||
expect(queues).toContain("rollup-full-backfill");
|
||||
expect(queues).toContain("system-job-alert");
|
||||
expect(queues).not.toContain("google-health-sync");
|
||||
});
|
||||
});
|
||||
@@ -30,10 +30,8 @@ import { prisma } from "@/lib/db";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { DENSE_INTRADAY_RETENTION_ENABLED } from "@/lib/jobs/dense-intraday-retention";
|
||||
import {
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
} from "@/lib/measurements/dense-intraday-retention";
|
||||
import { DENSE_INTRADAY_RETENTION_DAYS } from "@/lib/measurements/dense-intraday-retention";
|
||||
import { DENSE_INTRADAY_RETENTION_TYPES } from "@/lib/measurements/dense-intraday-sources";
|
||||
import { runDenseIntradayHourlyRebuild } from "@/lib/measurements/dense-intraday-hourly-rebuild";
|
||||
import { DENSE_INTRADAY_HOURLY_REBUILD_QUEUE } from "./dense-intraday-hourly-rebuild-queue";
|
||||
|
||||
|
||||
@@ -18,16 +18,19 @@
|
||||
* `src/lib/jobs/reminder-worker.ts` so pg-boss provisions it at boot; an
|
||||
* unregistered queue silently never drains.
|
||||
*/
|
||||
import { dropPendingDeletion } from "@/lib/account-deletion/active-account";
|
||||
import { ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import {
|
||||
runDenseIntradayRetention,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
DENSE_INTRADAY_FOLD_SOURCES,
|
||||
} from "@/lib/measurements/dense-intraday-retention";
|
||||
import {
|
||||
DENSE_INTRADAY_FOLD_SOURCES,
|
||||
DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
|
||||
} from "@/lib/measurements/dense-intraday-sources";
|
||||
import { isRollupFullBackfillPending } from "@/lib/rollups/full-backfill";
|
||||
import { DENSE_INTRADAY_RETENTION_QUEUE } from "./dense-intraday-retention-queue";
|
||||
|
||||
/**
|
||||
@@ -103,6 +106,7 @@ export async function runDenseIntradayRetentionForUser(
|
||||
const summary = await runDenseIntradayRetention(prisma, {
|
||||
userId,
|
||||
shouldStop,
|
||||
deferUser: isRollupFullBackfillPending,
|
||||
log: () => {
|
||||
// Silent inside the queue handler — the worker logs the totals.
|
||||
},
|
||||
@@ -133,8 +137,8 @@ export async function runDenseIntradayRetentionForUser(
|
||||
|
||||
/**
|
||||
* Boot-time discovery. Finds every user holding at least one LIVE
|
||||
* per-sample dense-tier row (a row of a `DENSE_INTRADAY_FOLD_SOURCES` source
|
||||
* and a `DENSE_INTRADAY_RETENTION_TYPES` type whose externalId does NOT start
|
||||
* per-sample dense-tier row (a row of a source and type the fold takes,
|
||||
* `DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE`, whose externalId does NOT start
|
||||
* with the daily-stats prefix, that is not tombstoned, AND whose
|
||||
* `measuredAt` is older than the retention window) and enqueues one
|
||||
* retention job per account.
|
||||
@@ -161,36 +165,40 @@ export async function enqueueBootTimeDenseIntradayRetention(): Promise<{
|
||||
}
|
||||
|
||||
try {
|
||||
// The dense-tier type set is a closed compile-time list of enum
|
||||
// members; splice-free — Prisma binds the `type IN (...)` array as
|
||||
// parameters.
|
||||
const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES);
|
||||
if (types.length === 0) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
}
|
||||
|
||||
const windowStart = new Date(
|
||||
Date.now() - DENSE_INTRADAY_RETENTION_DAYS * 24 * 60 * 60 * 1000,
|
||||
);
|
||||
|
||||
// `groupBy` compiles to a server-side `GROUP BY` — Prisma's
|
||||
// `distinct` dedups in the client AFTER pulling every matching row,
|
||||
// an instance-wide scan of every out-of-window dense-tier row just
|
||||
// to list user ids.
|
||||
const users = await dropPendingDeletion(
|
||||
prisma,
|
||||
await prisma.measurement.groupBy({
|
||||
by: ["userId"],
|
||||
where: {
|
||||
source: { in: [...DENSE_INTRADAY_FOLD_SOURCES] },
|
||||
type: { in: types },
|
||||
deletedAt: null,
|
||||
NOT: { externalId: { startsWith: "stats:" } },
|
||||
measuredAt: { lt: windowStart },
|
||||
},
|
||||
}),
|
||||
(u) => u.userId,
|
||||
);
|
||||
// One probe per account and source for the first raw sample the fold
|
||||
// would take, which stops at that row. A grouped count over every such
|
||||
// row read the whole out-of-window history of every account to list
|
||||
// their ids, and on an account with tens of millions of heart-rate rows
|
||||
// that is most of the table (#1297).
|
||||
const accounts = await prisma.user.findMany({
|
||||
where: ACTIVE_ACCOUNT,
|
||||
select: { id: true },
|
||||
orderBy: { createdAt: "asc" },
|
||||
});
|
||||
const users: Array<{ userId: string }> = [];
|
||||
for (const { id } of accounts) {
|
||||
for (const source of DENSE_INTRADAY_FOLD_SOURCES) {
|
||||
const raw = await prisma.measurement.findFirst({
|
||||
where: {
|
||||
userId: id,
|
||||
source,
|
||||
type: { in: [...DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE[source]] },
|
||||
deletedAt: null,
|
||||
NOT: { externalId: { startsWith: "stats:" } },
|
||||
measuredAt: { lt: windowStart },
|
||||
},
|
||||
select: { id: true },
|
||||
});
|
||||
if (raw) {
|
||||
users.push({ userId: id });
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (users.length === 0) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
|
||||
@@ -145,7 +145,7 @@ function explainTimeout(message: string, expireSeconds: number | null): string {
|
||||
* serialises to `{ message, stack, … }`; a thrown non-Error can be any JSON at
|
||||
* all, so the raw text is the fallback rather than an invented placeholder.
|
||||
*/
|
||||
function presentError(
|
||||
export function presentError(
|
||||
raw: string | null,
|
||||
expireSeconds: number | null = null,
|
||||
): string {
|
||||
|
||||
@@ -307,6 +307,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
|
||||
"summary_pending_healed",
|
||||
"suppressed_client_managed",
|
||||
"suppressed_discreet",
|
||||
// The hourly system job alert: queues failing now, notices sent, admins.
|
||||
"system_job_admins",
|
||||
"system_job_notices",
|
||||
"system_jobs_failing",
|
||||
"thumbnails_enqueued",
|
||||
"tls_pin_monitor_known_count",
|
||||
"tls_pin_monitor_outcome",
|
||||
@@ -330,6 +334,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
|
||||
"users_synced",
|
||||
"users_useful",
|
||||
"wedge_skipped",
|
||||
// The full rollup backfill: calendar-year windows one run folded.
|
||||
"windows_folded",
|
||||
]);
|
||||
|
||||
export const MAX_JOB_FACTS = 32;
|
||||
|
||||
@@ -116,12 +116,11 @@ import {
|
||||
hourOfDayForUserTz,
|
||||
} from "@/lib/measurements/consolidation-tz";
|
||||
import {
|
||||
DENSE_INTRADAY_FOLD_SOURCES,
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
deriveDailyRestingFromPulse,
|
||||
hourlyStatsExternalId,
|
||||
} from "@/lib/measurements/dense-intraday-retention";
|
||||
import { DENSE_INTRADAY_RETENTION_TYPES } from "@/lib/measurements/dense-intraday-sources";
|
||||
import {
|
||||
absorbIntoFold,
|
||||
foldConstituentHorizon,
|
||||
@@ -160,6 +159,9 @@ interface RepairStep {
|
||||
foldedAfterMs: number;
|
||||
}
|
||||
|
||||
/** Sources the dense fold took when this repair was written; later ones never split a window. */
|
||||
const REPAIRED_DENSE_SOURCES = ["APPLE_HEALTH", "HEALTH_CONNECT"] as const;
|
||||
|
||||
/** Fixed order, so a resume position (`step`) means the same on every run. */
|
||||
export const FOLD_REPAIR_STEPS: readonly RepairStep[] = [
|
||||
...[...HIGH_FREQUENCY_MEAN_TYPES].map((type): RepairStep => ({
|
||||
@@ -168,7 +170,7 @@ export const FOLD_REPAIR_STEPS: readonly RepairStep[] = [
|
||||
grain: "day",
|
||||
foldedAfterMs: CONSOLIDATION_GRACE_CUTOFF_HOURS * HOUR_MS,
|
||||
})),
|
||||
...DENSE_INTRADAY_FOLD_SOURCES.flatMap((source) =>
|
||||
...REPAIRED_DENSE_SOURCES.flatMap((source) =>
|
||||
[...DENSE_INTRADAY_RETENTION_TYPES].map((type): RepairStep => ({
|
||||
type,
|
||||
source,
|
||||
|
||||
@@ -51,7 +51,7 @@ import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-int
|
||||
import { HEALTH_CONNECT_IMPORT_SEND_OPTIONS } from "@/lib/jobs/health-connect-import-worker";
|
||||
import { MEASUREMENT_MAINTENANCE_SEND_OPTIONS } from "@/lib/jobs/measurement-maintenance";
|
||||
import { PULSE_DAY_MEAN_SEND_OPTIONS } from "@/lib/jobs/pulse-day-mean";
|
||||
import { ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS } from "@/lib/rollups/measurement-rollups";
|
||||
import { ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS } from "@/lib/rollups/full-backfill";
|
||||
import { MORNING_DIGEST_REFRESH_EXPIRE_SECONDS } from "@/lib/jobs/morning-digest-refresh-shared";
|
||||
import { INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS } from "@/lib/jobs/insight-status-generate-queue";
|
||||
import { WORKOUT_INSIGHT_GENERATE_EXPIRE_SECONDS } from "@/lib/jobs/workout-insight-generate-shared";
|
||||
@@ -261,7 +261,7 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
|
||||
fn: "handleRollupFullBackfill",
|
||||
},
|
||||
exclusive: "lockedPass",
|
||||
why: "One account's five-year fold: four aggregates under the statement timeout and four persist transactions of up to 120 s each, about thirteen minutes at worst under the defaults and unbounded when the operator lifts the statement timeout.",
|
||||
why: "One account's five-year fold, a calendar year at a time with every statement under its own five-minute limit; the run stops at its budget between windows and hands the remaining years to a follow-up.",
|
||||
},
|
||||
"mood-rollup-recompute": short("One bucket."),
|
||||
"mood-rollup-full-backfill": short(
|
||||
@@ -386,6 +386,9 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
|
||||
"recovery-readiness": short(
|
||||
"A few indexed reads, one row per account from two backup ledgers, and at most one notice per signal.",
|
||||
),
|
||||
"system-job-alert": short(
|
||||
"The newest fifty terminal runs of each system queue over pg-boss's completion index, and at most one notice per queue.",
|
||||
),
|
||||
"integration-watchdog-hourly": short(
|
||||
"A few indexed reads per account with a connection and no provider call; a 14-day history count only for a silent source.",
|
||||
),
|
||||
|
||||
@@ -56,9 +56,11 @@ import {
|
||||
import {
|
||||
ROLLUP_FULL_BACKFILL_CONCURRENCY,
|
||||
ROLLUP_FULL_BACKFILL_QUEUE,
|
||||
type RollupFullBackfillPayload,
|
||||
} from "@/lib/rollups/full-backfill";
|
||||
import {
|
||||
ROLLUP_RECOMPUTE_CONCURRENCY,
|
||||
ROLLUP_RECOMPUTE_QUEUE,
|
||||
type RollupFullBackfillPayload,
|
||||
} from "@/lib/rollups/measurement-rollups";
|
||||
import {
|
||||
MOOD_ROLLUP_FULL_BACKFILL_CONCURRENCY,
|
||||
@@ -213,7 +215,7 @@ const discovery: readonly BootDiscovery[] = [
|
||||
name: "rollup-full-backfill",
|
||||
run: async () =>
|
||||
(
|
||||
await import("@/lib/rollups/measurement-rollups")
|
||||
await import("@/lib/rollups/full-backfill")
|
||||
).enqueueBootTimeRollupBackfill(BOOT_BACKFILL_STAGGER_SECONDS * 0),
|
||||
},
|
||||
{
|
||||
|
||||
@@ -4,7 +4,8 @@
|
||||
* the morning digest refresh, the data-arrival spine with its reaction line
|
||||
* and workout insight, the nightly scores, the period-narrative warm, the
|
||||
* Coach memory refresh, nudge, reminder sweep and plan review, the low-stock
|
||||
* pass and the daily-briefing fallback.
|
||||
* pass, the daily-briefing fallback, and the hourly alert to the admins
|
||||
* about a system queue that keeps failing.
|
||||
*
|
||||
* Most queues here are keyless cron ticks, so they carry no policy. The
|
||||
* LLM-bound warm queues (pre-generate, period narrative, Coach memory) pass
|
||||
@@ -77,6 +78,12 @@ import {
|
||||
WORKOUT_INSIGHT_GENERATE_QUEUE,
|
||||
} from "@/lib/jobs/workout-insight-generate-shared";
|
||||
|
||||
import { cronIsTheRetry } from "@/lib/jobs/reminder/registrar-shared";
|
||||
import {
|
||||
SYSTEM_JOB_ALERT_CRON,
|
||||
SYSTEM_JOB_ALERT_QUEUE,
|
||||
} from "@/lib/jobs/system-job-alert-queue";
|
||||
|
||||
import { defineJob, type JobDefinition, type JobGroup } from "./define-job";
|
||||
|
||||
const GENERAL_STATUS_QUEUE = "insights-general-status";
|
||||
@@ -261,6 +268,18 @@ const jobs: readonly JobDefinition[] = [
|
||||
work: { localConcurrency: 1 },
|
||||
load: async () => (await handlers()).handleDailyBriefing,
|
||||
}),
|
||||
// Hourly: tell the admins about a system queue that keeps failing, the way
|
||||
// the recovery readiness evaluation tells them about a backup. Single-flight
|
||||
// and no retries; the next tick is the retry, and the notice claim keeps
|
||||
// two evaluations from sending the same notice.
|
||||
defineJob({
|
||||
queue: SYSTEM_JOB_ALERT_QUEUE,
|
||||
cron: SYSTEM_JOB_ALERT_CRON,
|
||||
scheduleOptions: cronIsTheRetry,
|
||||
work: { localConcurrency: 1 },
|
||||
load: async () =>
|
||||
(await import("@/lib/jobs/system-job-alert")).handleSystemJobAlert,
|
||||
}),
|
||||
];
|
||||
|
||||
export const STATUS_GROUP: JobGroup = {
|
||||
|
||||
@@ -56,8 +56,14 @@ import {
|
||||
type CumulativePrRederivePayload,
|
||||
} from "@/lib/personal-records/cumulative-pr-rederivation";
|
||||
import {
|
||||
recomputeUserRollups,
|
||||
foldUserRollupHistory,
|
||||
isRollupFullBackfillPending,
|
||||
ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
sendRollupFullBackfillContinuation,
|
||||
type RollupFullBackfillPayload,
|
||||
} from "@/lib/rollups/full-backfill";
|
||||
import {
|
||||
recomputeUserRollups,
|
||||
type RollupRecomputePayload,
|
||||
} from "@/lib/rollups/measurement-rollups";
|
||||
import {
|
||||
@@ -91,35 +97,47 @@ export async function handleRollupRecompute(
|
||||
}
|
||||
|
||||
/**
|
||||
* The full fold of one uncovered account across the default window and every
|
||||
* granularity. Each granularity commits on its own, so a run stopped on its
|
||||
* budget throws and the retry folds the rest.
|
||||
* The full fold of one account across the five-year window, a calendar year
|
||||
* at a time and every statement under the full backfill's own limit (#1298).
|
||||
* A run that reaches its job budget hands the years it did not reach to a
|
||||
* follow-up and completes; only a run that could not finish even one window
|
||||
* fails, and its retry starts that window again.
|
||||
*/
|
||||
export async function handleRollupFullBackfill(
|
||||
jobs: Job<RollupFullBackfillPayload>[],
|
||||
): Promise<JobOutcome> {
|
||||
const shouldStop = jobBudget(jobs);
|
||||
let rowsUpsertedTotal = 0;
|
||||
let windowsTotal = 0;
|
||||
let continued = 0;
|
||||
for (const job of jobs) {
|
||||
const { userId } = job.data;
|
||||
const { rowsUpserted, durationMs, stopped } = await recomputeUserRollups(
|
||||
userId,
|
||||
{ shouldStop },
|
||||
);
|
||||
if (stopped) {
|
||||
throw new Error(
|
||||
`rollup-full-backfill for user ${userId} stopped on its job budget; the retry folds the rest`,
|
||||
);
|
||||
const { userId, resumeBefore } = job.data;
|
||||
const done = await foldUserRollupHistory(userId, {
|
||||
before: resumeBefore ? new Date(resumeBefore) : undefined,
|
||||
shouldStop,
|
||||
statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
});
|
||||
rowsUpsertedTotal += done.rowsUpserted;
|
||||
windowsTotal += done.windowsFolded;
|
||||
if (done.resumeBefore) {
|
||||
if (done.windowsFolded === 0) {
|
||||
throw new Error(
|
||||
`rollup-full-backfill for user ${userId} stopped on its job budget before it folded a window; the retry starts that window again`,
|
||||
);
|
||||
}
|
||||
await sendRollupFullBackfillContinuation(userId, done.resumeBefore);
|
||||
continued += 1;
|
||||
}
|
||||
rowsUpsertedTotal += rowsUpserted;
|
||||
workerLog(
|
||||
"info",
|
||||
`[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
|
||||
`[rollup-full-backfill] user=${userId} rows=${done.rowsUpserted} windows=${done.windowsFolded} duration=${done.durationMs}ms${done.resumeBefore ? ` continuesBefore=${done.resumeBefore.toISOString()}` : ""}`,
|
||||
);
|
||||
}
|
||||
return jobDone({
|
||||
users: jobs.length,
|
||||
rows_upserted: rowsUpsertedTotal,
|
||||
windows_folded: windowsTotal,
|
||||
continued,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -510,6 +528,7 @@ export function drainCumulativeHandler(
|
||||
{
|
||||
dryRun: false,
|
||||
shouldStop,
|
||||
deferUser: isRollupFullBackfillPending,
|
||||
log: (line) => workerLog("info", line),
|
||||
},
|
||||
);
|
||||
|
||||
@@ -67,6 +67,7 @@ const subsystemSurface = {
|
||||
"coach-plan-review": { audience: "system" },
|
||||
"medication-low-stock": { audience: "system" },
|
||||
"daily-briefing": { audience: "system" },
|
||||
"system-job-alert": { audience: "system" },
|
||||
|
||||
// register-rollup.ts
|
||||
"rollup-recompute": { audience: "system" },
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
/**
|
||||
* The queue facts of `system-job-alert.ts`: its name and schedule. They live
|
||||
* apart from the job so the worker registry can read them without loading it.
|
||||
*/
|
||||
export const SYSTEM_JOB_ALERT_QUEUE = "system-job-alert";
|
||||
|
||||
/**
|
||||
* Hourly at :43, a minute no other schedule uses. A notice goes out once per
|
||||
* change whatever the cadence, and an hour is soon enough to hear that a
|
||||
* background job keeps failing.
|
||||
*/
|
||||
export const SYSTEM_JOB_ALERT_CRON = "43 * * * *";
|
||||
@@ -0,0 +1,303 @@
|
||||
/**
|
||||
* Push to every admin when a background job of the instance keeps failing:
|
||||
* once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` runs in a row, a
|
||||
* reminder at most once a day while it keeps failing, and once when it runs
|
||||
* successfully again. Nothing else.
|
||||
*
|
||||
* Why (#1298). A full rollup backfill that failed on every boot showed only
|
||||
* on the admin page's card of failing background jobs. From the account's
|
||||
* side the only symptom was empty dashboard cards, and nothing pointed an
|
||||
* operator at the card. The recovery readiness evaluation already pushes a
|
||||
* stopped backup; this does the same for the queues whose failure has no
|
||||
* account-facing status of its own (the `system` audience in
|
||||
* `subsystem-surface.ts`).
|
||||
*
|
||||
* The streak is read from pg-boss's own ledger, newest terminal run first: a
|
||||
* run that failed (or completed and said every candidate failed,
|
||||
* `did.all_failed`) adds its attempts, retries included, and the first run
|
||||
* that succeeded ends the streak. A queue with no terminal run left in the
|
||||
* ledger is not judged at all.
|
||||
*
|
||||
* The memory is one row per announced queue (`SystemJobAlertNotice`), claimed
|
||||
* with a compare-and-set before the notice is sent, exactly as the readiness
|
||||
* notices are (`readiness-notify.ts`, whose decision this reuses), so two
|
||||
* workers send a notice once and a crash between the claim and the send loses
|
||||
* a notice rather than repeating it. Delivery is the ordinary localised path,
|
||||
* as the other operator alerts use: a `SYSTEM_ALERT` to every admin, in their
|
||||
* locale, on the channels they enabled.
|
||||
*/
|
||||
import type { Job } from "pg-boss";
|
||||
|
||||
import { ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { presentError } from "@/lib/jobs/job-failures";
|
||||
import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
|
||||
import { SUBSYSTEM_SURFACE } from "@/lib/jobs/subsystem-surface";
|
||||
import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
import { dispatchLocalisedNotification } from "@/lib/notifications/dispatch-localised";
|
||||
import {
|
||||
decideNotice,
|
||||
type NoticeKind,
|
||||
type StoredNotice,
|
||||
} from "@/lib/recovery/readiness-notify";
|
||||
|
||||
/** Failed runs in a row, retries included, before the admins are told. */
|
||||
export const SYSTEM_JOB_FAILURE_STREAK = 3;
|
||||
|
||||
/** Newest terminal runs read per queue. */
|
||||
const LOOKBACK_ROWS = 50;
|
||||
|
||||
/** Where the notice takes the admin: the page with the failing-jobs card. */
|
||||
const ADMIN_STATUS_PATH = "/admin/system-status";
|
||||
|
||||
/** One terminal run as the ledger holds it, newest first. */
|
||||
export interface TerminalRun {
|
||||
state: "completed" | "failed";
|
||||
/** Retries pg-boss ran before this run settled. */
|
||||
retryCount: number;
|
||||
/** The run completed but reported that every candidate failed. */
|
||||
allFailed: boolean;
|
||||
completedOn: Date;
|
||||
error: string | null;
|
||||
expireSeconds: number | null;
|
||||
}
|
||||
|
||||
/** What the ledger says about one queue. */
|
||||
export interface QueueVerdict {
|
||||
/**
|
||||
* `failing`: the streak reached the threshold. `ok`: the newest run, or a
|
||||
* run after the last failures, succeeded. `pending`: fewer failures than
|
||||
* the threshold and no success to judge by. `inactive`: no run left.
|
||||
*/
|
||||
state: "failing" | "ok" | "pending" | "inactive";
|
||||
/** Failed attempts in a row, newest first. */
|
||||
failures: number;
|
||||
/** The newest failure's message, redacted and truncated. */
|
||||
lastError: string;
|
||||
}
|
||||
|
||||
/** Pure: judge one queue's newest terminal runs. */
|
||||
export function judgeQueueRuns(runs: readonly TerminalRun[]): QueueVerdict {
|
||||
if (runs.length === 0)
|
||||
return { state: "inactive", failures: 0, lastError: "" };
|
||||
let failures = 0;
|
||||
let lastError: string | null = null;
|
||||
let succeeded = false;
|
||||
for (const run of runs) {
|
||||
if (run.state === "completed" && !run.allFailed) {
|
||||
succeeded = true;
|
||||
break;
|
||||
}
|
||||
failures += run.retryCount + 1;
|
||||
lastError ??= presentError(run.error, run.expireSeconds);
|
||||
}
|
||||
const state =
|
||||
failures >= SYSTEM_JOB_FAILURE_STREAK
|
||||
? "failing"
|
||||
: succeeded
|
||||
? "ok"
|
||||
: "pending";
|
||||
return { state, failures, lastError: lastError ?? "" };
|
||||
}
|
||||
|
||||
/** The queues this alert watches: every one whose failure has no other surface. */
|
||||
export function watchedQueues(): string[] {
|
||||
return Object.entries(SUBSYSTEM_SURFACE)
|
||||
.filter(([, surface]) => surface.audience === "system")
|
||||
.map(([queue]) => queue);
|
||||
}
|
||||
|
||||
/**
|
||||
* The newest terminal runs of each queue, newest first, one statement over
|
||||
* pg-boss's `(name, completed_on)` index. `null` when the queue schema is not
|
||||
* there (a web-only deployment): no queue to ask is not a healthy queue.
|
||||
*/
|
||||
export async function readTerminalRuns(
|
||||
queues: readonly string[],
|
||||
): Promise<Map<string, TerminalRun[]> | null> {
|
||||
try {
|
||||
const rows = await prisma.$queryRaw<
|
||||
Array<{
|
||||
name: string;
|
||||
state: string;
|
||||
retry_count: number;
|
||||
all_failed: string | null;
|
||||
completed_on: Date;
|
||||
error: string | null;
|
||||
expire_seconds: number | null;
|
||||
}>
|
||||
>`
|
||||
SELECT q.name, j.state, j.retry_count, j.all_failed, j.completed_on,
|
||||
j.error, j.expire_seconds
|
||||
FROM unnest(${queues as string[]}::text[]) AS q(name)
|
||||
CROSS JOIN LATERAL (
|
||||
SELECT state, retry_count, completed_on, expire_seconds,
|
||||
output->'did'->>'all_failed' AS all_failed,
|
||||
COALESCE(
|
||||
output->>'message',
|
||||
output->'value'->>'message',
|
||||
output::text
|
||||
) AS error
|
||||
FROM pgboss.job
|
||||
WHERE name = q.name
|
||||
AND completed_on IS NOT NULL
|
||||
AND state IN ('completed', 'failed')
|
||||
ORDER BY completed_on DESC
|
||||
LIMIT ${LOOKBACK_ROWS}
|
||||
) j
|
||||
ORDER BY q.name, j.completed_on DESC
|
||||
`;
|
||||
const byQueue = new Map<string, TerminalRun[]>();
|
||||
for (const row of rows) {
|
||||
const runs = byQueue.get(row.name) ?? [];
|
||||
runs.push({
|
||||
state: row.state === "failed" ? "failed" : "completed",
|
||||
retryCount: Number(row.retry_count ?? 0),
|
||||
allFailed: row.all_failed === "true",
|
||||
completedOn: row.completed_on,
|
||||
error: row.error,
|
||||
expireSeconds: row.expire_seconds,
|
||||
});
|
||||
byQueue.set(row.name, runs);
|
||||
}
|
||||
return byQueue;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function titleKey(kind: NoticeKind): string {
|
||||
if (kind === "recovered")
|
||||
return "notifications.systemJobAlert.titleRecovered";
|
||||
if (kind === "reminder") return "notifications.systemJobAlert.titleReminder";
|
||||
return "notifications.systemJobAlert.titleFailing";
|
||||
}
|
||||
|
||||
export interface SystemJobAlertOutcome {
|
||||
/** Queues at or past the threshold now. */
|
||||
failing: number;
|
||||
/** Notices claimed and sent, by queue. */
|
||||
sent: Array<{ queue: string; kind: NoticeKind }>;
|
||||
/** Admins each notice went to. */
|
||||
admins: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Judge every watched queue against what the admins were last told, claim
|
||||
* every notice that is due, and send it.
|
||||
*/
|
||||
export async function notifySystemJobFailures(
|
||||
now: Date = new Date(),
|
||||
): Promise<SystemJobAlertOutcome> {
|
||||
const outcome: SystemJobAlertOutcome = { failing: 0, sent: [], admins: 0 };
|
||||
const runs = await readTerminalRuns(watchedQueues());
|
||||
if (runs === null) return outcome;
|
||||
|
||||
// account-deletion: one row per queue of the instance; no account in it.
|
||||
const rows = await prisma.systemJobAlertNotice.findMany();
|
||||
const byQueue = new Map(rows.map((row) => [row.queue, row]));
|
||||
const admins = await prisma.user.findMany({
|
||||
where: { role: "ADMIN", ...ACTIVE_ACCOUNT },
|
||||
select: { id: true },
|
||||
});
|
||||
outcome.admins = admins.length;
|
||||
|
||||
for (const queue of watchedQueues()) {
|
||||
const verdict = judgeQueueRuns(runs.get(queue) ?? []);
|
||||
if (verdict.state === "failing") outcome.failing += 1;
|
||||
// Too few failures to call it a streak and no success to call it over.
|
||||
if (verdict.state === "pending") continue;
|
||||
|
||||
const row = byQueue.get(queue);
|
||||
const stored: StoredNotice | null = row
|
||||
? {
|
||||
notifiedState:
|
||||
row.notifiedState === "failing" || row.notifiedState === "ok"
|
||||
? row.notifiedState
|
||||
: null,
|
||||
notifiedAt: row.notifiedAt,
|
||||
}
|
||||
: null;
|
||||
const decision = decideNotice(stored, verdict.state, now);
|
||||
// With no admin yet there is nobody to tell, so nothing is claimed: the
|
||||
// first admin hears about a queue that is still failing.
|
||||
if (!decision.next || admins.length === 0) continue;
|
||||
|
||||
// Claim: only the evaluation that moves the row from what it read sends.
|
||||
const claimed = row
|
||||
? // account-deletion: the one queue row read above; no account in it.
|
||||
(
|
||||
await prisma.systemJobAlertNotice.updateMany({
|
||||
where: {
|
||||
queue,
|
||||
notifiedState: row.notifiedState,
|
||||
notifiedAt: row.notifiedAt,
|
||||
},
|
||||
data: {
|
||||
notifiedState: decision.next.notifiedState,
|
||||
notifiedAt: decision.next.notifiedAt,
|
||||
},
|
||||
})
|
||||
).count
|
||||
: (
|
||||
await prisma.systemJobAlertNotice.createMany({
|
||||
data: [
|
||||
{
|
||||
queue,
|
||||
notifiedState: decision.next.notifiedState,
|
||||
notifiedAt: decision.next.notifiedAt,
|
||||
},
|
||||
],
|
||||
skipDuplicates: true,
|
||||
})
|
||||
).count;
|
||||
if (claimed !== 1 || decision.send === null) continue;
|
||||
|
||||
for (const admin of admins) {
|
||||
await dispatchLocalisedNotification({
|
||||
userId: admin.id,
|
||||
eventType: "SYSTEM_ALERT",
|
||||
titleKey: titleKey(decision.send),
|
||||
messageKey:
|
||||
decision.send === "recovered"
|
||||
? "notifications.systemJobAlert.bodyRecovered"
|
||||
: "notifications.systemJobAlert.bodyFailing",
|
||||
params: {
|
||||
queue,
|
||||
failures: verdict.failures,
|
||||
error: verdict.lastError,
|
||||
},
|
||||
metadata: {
|
||||
source: "system-job-alert",
|
||||
url: ADMIN_STATUS_PATH,
|
||||
queue,
|
||||
kind: decision.send,
|
||||
},
|
||||
});
|
||||
}
|
||||
outcome.sent.push({ queue, kind: decision.send });
|
||||
}
|
||||
return outcome;
|
||||
}
|
||||
|
||||
/** The hourly evaluation. Single-flight; the next tick is the retry. */
|
||||
export async function handleSystemJobAlert(
|
||||
jobs: Job<object>[],
|
||||
): Promise<JobOutcome> {
|
||||
void jobs;
|
||||
return withBackgroundEvent("job.system_job_alert", async (evt) => {
|
||||
try {
|
||||
const outcome = await notifySystemJobFailures();
|
||||
const did = {
|
||||
system_jobs_failing: outcome.failing,
|
||||
system_job_notices: outcome.sent.length,
|
||||
system_job_admins: outcome.admins,
|
||||
};
|
||||
for (const [key, value] of Object.entries(did)) evt.addMeta(key, value);
|
||||
return jobDone(did);
|
||||
} catch (err) {
|
||||
evt.addWarning(`system-job-alert failed: ${err}`);
|
||||
return jobFailed("system job alert evaluation failed", err);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -26,7 +26,7 @@ import {
|
||||
CUMULATIVE_HK_TYPES,
|
||||
HIGH_FREQUENCY_MEAN_TYPES,
|
||||
} from "../apple-health-mapping";
|
||||
import { DENSE_INTRADAY_RETENTION_TYPES } from "../dense-intraday-retention";
|
||||
import { DENSE_INTRADAY_RETENTION_TYPES } from "../dense-intraday-sources";
|
||||
|
||||
describe("dense intra-day retention — drain exemption (WX-E)", () => {
|
||||
it("EXEMPTS HEART_RATE_VARIABILITY from the destructive daily-mean drain", () => {
|
||||
|
||||
@@ -21,8 +21,8 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
|
||||
import {
|
||||
runDenseIntradayRetention,
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
} from "../dense-intraday-retention";
|
||||
import { DENSE_INTRADAY_RETENTION_TYPES } from "../dense-intraday-sources";
|
||||
import type { PerSampleRow } from "../drain-per-sample-cumulative";
|
||||
import { foldBoundary } from "../consolidation-tz";
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
@@ -161,7 +161,17 @@ describe("runDenseIntradayRetention — retention bound", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("scans exactly the dense-tier types, once per fold source, Apple Health first", async () => {
|
||||
it("leaves an account alone while its rollup tier is being rebuilt", async () => {
|
||||
// A full rollup backfill of the account still has to read the raw
|
||||
// samples; folding them first would leave it only hourly means (#1298).
|
||||
const { mock, findManyMeasurement } = buildPrismaMock({});
|
||||
const deferUser = vi.fn(async () => true);
|
||||
await runDenseIntradayRetention(mock, { deferUser, log: () => {} });
|
||||
expect(deferUser).toHaveBeenCalled();
|
||||
expect(findManyMeasurement).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("scans each fold source's dense types, once per source, Apple Health first", async () => {
|
||||
const { mock, findManyMeasurement } = buildPrismaMock({});
|
||||
await runDenseIntradayRetention(mock, { log: () => {} });
|
||||
|
||||
@@ -172,9 +182,13 @@ describe("runDenseIntradayRetention — retention bound", () => {
|
||||
const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES);
|
||||
// One walk per source, each scoped to its one source: two sources
|
||||
// never share a mean, and manual or Withings rows are never scanned.
|
||||
// The provider syncs fold only what they store sample by sample: heart
|
||||
// rate, never their daily SpO2 (#1297).
|
||||
expect(scans).toEqual([
|
||||
...types.map((type) => `APPLE_HEALTH:${type}`),
|
||||
...types.map((type) => `HEALTH_CONNECT:${type}`),
|
||||
"GOOGLE_HEALTH:PULSE",
|
||||
"OURA:PULSE",
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -14,6 +14,7 @@ import {
|
||||
findCompactionTombstones,
|
||||
findFoldedWindowDuplicates,
|
||||
isFoldedWindowCandidate,
|
||||
withoutFoldedSamples,
|
||||
} from "../folded-window";
|
||||
|
||||
const DAY = 86_400_000;
|
||||
@@ -332,3 +333,95 @@ describe("Health Connect", () => {
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("provider syncs (#1297)", () => {
|
||||
const old = new Date("2026-03-01T10:15:00.000Z");
|
||||
const hour = "stats:HKQuantityTypeIdentifierHeartRate:2026-03-01T10";
|
||||
|
||||
it("takes the heart rate Google Health and Oura store per sample, and nothing else of theirs", () => {
|
||||
for (const source of ["GOOGLE_HEALTH", "OURA"]) {
|
||||
expect(isFoldedWindowCandidate(raw("PULSE", old, { source }), NOW)).toBe(
|
||||
true,
|
||||
);
|
||||
// A nightly or daily SpO2 value is no intraday stream: never folded.
|
||||
expect(
|
||||
isFoldedWindowCandidate(raw("OXYGEN_SATURATION", old, { source }), NOW),
|
||||
).toBe(false);
|
||||
}
|
||||
// Neither stores intraday heart rate as samples.
|
||||
for (const source of ["FITBIT", "WHOOP", "WITHINGS"]) {
|
||||
expect(isFoldedWindowCandidate(raw("PULSE", old, { source }), NOW)).toBe(
|
||||
false,
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
it("drops the samples of an hour the fold took, under the write's own source", async () => {
|
||||
// The externalId read finds the hour's row; the reach read finds nothing.
|
||||
const findMany = vi.fn(
|
||||
async (args: {
|
||||
where: { source: string; externalId: { in?: string[] } };
|
||||
}) =>
|
||||
args.where.source === "GOOGLE_HEALTH" &&
|
||||
args.where.externalId.in?.includes(hour)
|
||||
? [{ type: "PULSE", externalId: hour }]
|
||||
: [],
|
||||
);
|
||||
const findUnique = vi.fn(async () => ({ timezone: "UTC" }));
|
||||
const rows = [
|
||||
{ type: "PULSE", externalId: "hr:a", measuredAt: old, value: 61 },
|
||||
{
|
||||
type: "PULSE",
|
||||
externalId: "hr:b",
|
||||
measuredAt: new Date("2026-03-01T12:15:00.000Z"),
|
||||
value: 62,
|
||||
},
|
||||
{
|
||||
type: "PULSE",
|
||||
externalId: "hr:c",
|
||||
measuredAt: new Date(NOW.getTime() - DAY),
|
||||
value: 63,
|
||||
},
|
||||
];
|
||||
const { kept, dropped } = await withoutFoldedSamples(
|
||||
{
|
||||
measurement: { findMany },
|
||||
user: { findUnique },
|
||||
} as unknown as Parameters<typeof withoutFoldedSamples>[0],
|
||||
"u1",
|
||||
"GOOGLE_HEALTH",
|
||||
rows,
|
||||
{ now: NOW },
|
||||
);
|
||||
expect(dropped).toBe(1);
|
||||
expect(kept.map((row) => row.externalId)).toEqual(["hr:b", "hr:c"]);
|
||||
for (const call of findMany.mock.calls) {
|
||||
expect(call[0].where.source).toBe("GOOGLE_HEALTH");
|
||||
}
|
||||
});
|
||||
|
||||
it("asks nothing for a write of recent samples", async () => {
|
||||
const findMany = vi.fn();
|
||||
const findUnique = vi.fn();
|
||||
const rows = [
|
||||
{
|
||||
type: "PULSE",
|
||||
externalId: "heartrate:x",
|
||||
measuredAt: new Date(NOW.getTime() - 2 * 3_600_000),
|
||||
},
|
||||
];
|
||||
const { kept } = await withoutFoldedSamples(
|
||||
{
|
||||
measurement: { findMany },
|
||||
user: { findUnique },
|
||||
} as unknown as Parameters<typeof withoutFoldedSamples>[0],
|
||||
"u1",
|
||||
"OURA",
|
||||
rows,
|
||||
{ now: NOW },
|
||||
);
|
||||
expect(kept).toEqual(rows);
|
||||
expect(findMany).not.toHaveBeenCalled();
|
||||
expect(findUnique).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -76,6 +76,14 @@ export interface ConsolidationOptions {
|
||||
* day this one did not reach.
|
||||
*/
|
||||
shouldStop?: () => boolean;
|
||||
/**
|
||||
* Asked once per account before it is walked; `true` leaves the account
|
||||
* for a later run, and nothing of it is read or written. The dense fold
|
||||
* passes `isRollupFullBackfillPending`: while an account's rollup tier is
|
||||
* rebuilt from its raw readings, folding those readings would leave the
|
||||
* rebuild only hourly means to take the old days' figures from.
|
||||
*/
|
||||
deferUser?: (userId: string) => Promise<boolean>;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -458,6 +466,10 @@ export async function runConsolidation<TType extends MeasurementType>(
|
||||
const users = await loadConsolidationUsers(prismaClient, options.userId);
|
||||
|
||||
walk: for (const user of users) {
|
||||
if (await options.deferUser?.(user.id)) {
|
||||
annotate({ meta: { consolidation_deferred_user: true } });
|
||||
continue;
|
||||
}
|
||||
// The zone the server publishes for the account (`/api/auth/me`), which
|
||||
// is the zone the phone cuts its `stats:` days and hour buckets in and
|
||||
// the one the hour-bucket writer stores them under. A stored value the
|
||||
|
||||
@@ -72,11 +72,11 @@ import { localDayWindow } from "@/lib/tz/local-day";
|
||||
import { meanBucketValue } from "./consolidate-daily-mean";
|
||||
import {
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
adoptOrMintHourlyRows,
|
||||
bucketRowsByLocalHour,
|
||||
hourlyStatsExternalId,
|
||||
} from "./dense-intraday-retention";
|
||||
import { DENSE_INTRADAY_RETENTION_TYPES } from "./dense-intraday-sources";
|
||||
import {
|
||||
bucketSpan,
|
||||
enqueueRollupRecompute,
|
||||
|
||||
@@ -56,7 +56,6 @@ import {
|
||||
FOLD_TRANSACTION_OPTIONS,
|
||||
} from "@/lib/export/restore-lock";
|
||||
import type {
|
||||
MeasurementSource,
|
||||
MeasurementType,
|
||||
Prisma,
|
||||
PrismaClient,
|
||||
@@ -67,7 +66,11 @@ import {
|
||||
dailyStatsExternalId,
|
||||
hkIdentifierForType,
|
||||
} from "./apple-health-mapping";
|
||||
import { runConsolidation, type DayWriteOutcome } from "./consolidation-base";
|
||||
import {
|
||||
runConsolidation,
|
||||
type ConsolidationOptions,
|
||||
type DayWriteOutcome,
|
||||
} from "./consolidation-base";
|
||||
import {
|
||||
canonicalHourlyTimestamp,
|
||||
hourOfDayForUserTz,
|
||||
@@ -78,6 +81,11 @@ import { percentile } from "@/lib/insights/strain-score";
|
||||
import type { PerSampleRow } from "./drain-per-sample-cumulative";
|
||||
import { DEFAULT_TIMEZONE } from "@/lib/tz/format";
|
||||
import { localDayWindow } from "@/lib/tz/local-day";
|
||||
import {
|
||||
DENSE_INTRADAY_FOLD_SOURCES,
|
||||
DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
|
||||
type DenseIntradayFoldSource,
|
||||
} from "./dense-intraday-sources";
|
||||
import {
|
||||
absorbIntoFold,
|
||||
foldConstituentHorizon,
|
||||
@@ -121,40 +129,6 @@ import {
|
||||
* they did against the raw stream's successor rows before.
|
||||
*/
|
||||
|
||||
/**
|
||||
* The dense-tier types. These are the high-frequency intra-day signals that
|
||||
* accumulate raw per-sample rows the destructive daily-mean drain never
|
||||
* touches:
|
||||
*
|
||||
* - `HEART_RATE_VARIABILITY` + `PULSE` — the daytime shape the Stress
|
||||
* engine reads (v1.10.0 / v1.18.11).
|
||||
* - `OXYGEN_SATURATION` — Apple Watch's Blood-Oxygen app samples SpO2
|
||||
* periodically through the day and overnight; the readings map to
|
||||
* `aggregation: "latest"` and sit in NEITHER `CUMULATIVE_HK_TYPES` nor
|
||||
* `HIGH_FREQUENCY_MEAN_TYPES`, so every sample piled up raw forever.
|
||||
* Folded here to hourly-mean rows with the daily min/max preserved on
|
||||
* the rollup tier (the lowest sample is the overnight-desaturation
|
||||
* nadir — clinically the meaningful figure, not the mean).
|
||||
*
|
||||
* Every member is deliberately EXEMPT from the destructive daily-mean drain
|
||||
* (`HIGH_FREQUENCY_MEAN_TYPES`); the disjointness from that set is the
|
||||
* load-bearing invariant a regression test pins — a type in both would be
|
||||
* folded by two drains at once and corrupt the value.
|
||||
*
|
||||
* `RESPIRATORY_RATE` is deliberately NOT here: it is already a member of
|
||||
* `HIGH_FREQUENCY_MEAN_TYPES`, so the nightly mean-consolidation already
|
||||
* collapses it to a daily-mean row. Adding it here would double-fold it.
|
||||
*
|
||||
* One fold grain per tier: every member folds HOURLY. A per-type grain
|
||||
* split would fork the write path and the rebuild for no reader benefit.
|
||||
*/
|
||||
export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
|
||||
new Set<MeasurementType>([
|
||||
"HEART_RATE_VARIABILITY",
|
||||
"PULSE",
|
||||
"OXYGEN_SATURATION",
|
||||
]);
|
||||
|
||||
/**
|
||||
* Retention bound (days). Per-sample dense-tier rows older than this window
|
||||
* fold to hourly-mean rows and the raw rows are deleted; rows inside the
|
||||
@@ -173,23 +147,6 @@ export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
|
||||
*/
|
||||
export const DENSE_INTRADAY_RETENTION_DAYS = 90;
|
||||
|
||||
/**
|
||||
* The sources whose raw dense samples fold (v1.42). Apple Health from the
|
||||
* start; Health Connect because its importer writes the samples of the last
|
||||
* 90 days raw, exactly like an Apple Health sync, and without a fold every
|
||||
* new export would add another quarter of minute-by-minute heart rate that
|
||||
* never ages into hourly means. Each source folds into its OWN `stats:` rows,
|
||||
* so an hour two sources both covered keeps one mean per source, the same
|
||||
* way the raw samples kept their source.
|
||||
*/
|
||||
export const DENSE_INTRADAY_FOLD_SOURCES = [
|
||||
"APPLE_HEALTH",
|
||||
"HEALTH_CONNECT",
|
||||
] as const satisfies readonly MeasurementSource[];
|
||||
|
||||
export type DenseIntradayFoldSource =
|
||||
(typeof DENSE_INTRADAY_FOLD_SOURCES)[number];
|
||||
|
||||
/** The stats externalId prefix marks an already-collapsed row. */
|
||||
const DAILY_STATS_PREFIX = "stats:";
|
||||
|
||||
@@ -485,13 +442,8 @@ export interface DenseIntradayRetentionSummary {
|
||||
};
|
||||
}
|
||||
|
||||
export interface DenseIntradayRetentionOptions {
|
||||
/** Limit the pass to a single user. Default = every user. */
|
||||
userId?: string;
|
||||
/** Preview-only mode — no DB writes. */
|
||||
dryRun?: boolean;
|
||||
/** Logger sink — defaults to `console.log`. */
|
||||
log?: (line: string) => void;
|
||||
/** The shared drain options (`ConsolidationOptions`) and the retention window. */
|
||||
export interface DenseIntradayRetentionOptions extends ConsolidationOptions {
|
||||
/**
|
||||
* Retention window in days. Per-sample rows OLDER than `now() -
|
||||
* retentionDays` are folded; rows inside the window stay raw. Defaults to
|
||||
@@ -500,11 +452,6 @@ export interface DenseIntradayRetentionOptions {
|
||||
* full-collapse, never by the scheduled pass.
|
||||
*/
|
||||
retentionDays?: number;
|
||||
/**
|
||||
* Ends the pass cleanly before the next day when it returns `true`; the
|
||||
* summary then reports `stoppedEarly`. See `ConsolidationOptions`.
|
||||
*/
|
||||
shouldStop?: () => boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -590,7 +537,7 @@ export async function runDenseIntradayRetention(
|
||||
await runConsolidation<MeasurementType>({
|
||||
prismaClient,
|
||||
options: { ...options, cutoffHours },
|
||||
types: DENSE_INTRADAY_RETENTION_TYPES,
|
||||
types: DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE[source],
|
||||
hkIdentifierForType,
|
||||
dailyStatsExternalId,
|
||||
statsPrefix: DAILY_STATS_PREFIX,
|
||||
@@ -1034,8 +981,8 @@ export async function runDenseIntradayRetention(
|
||||
},
|
||||
});
|
||||
|
||||
// One walk per source, Apple Health first so its behaviour and its order
|
||||
// are exactly what they were before Health Connect folded too. A walk that
|
||||
// One walk per source, in the map's order (Apple Health first, so its
|
||||
// behaviour is what it was before other sources folded too). A walk that
|
||||
// stopped early stops the pass: the next run resumes where it left off.
|
||||
for (const source of DENSE_INTRADAY_FOLD_SOURCES) {
|
||||
const walk = await walkSource(source);
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
/**
|
||||
* What the dense hourly fold (`dense-intraday-retention.ts`) takes: the
|
||||
* dense-tier types, and for each source the types it stores one row per
|
||||
* intraday sample of. The `folded_window` ingest guard (`folded-window.ts`)
|
||||
* reads the same map, so a source's raw samples are refused at ingest for
|
||||
* exactly the hours the fold has taken from it.
|
||||
*/
|
||||
import type {
|
||||
MeasurementSource,
|
||||
MeasurementType,
|
||||
} from "@/generated/prisma/client";
|
||||
|
||||
/**
|
||||
* The dense-tier types. These are the high-frequency intra-day signals that
|
||||
* accumulate raw per-sample rows the destructive daily-mean drain never
|
||||
* touches:
|
||||
*
|
||||
* - `HEART_RATE_VARIABILITY` + `PULSE` — the daytime shape the Stress
|
||||
* engine reads.
|
||||
* - `OXYGEN_SATURATION` — Apple Watch's Blood-Oxygen app samples SpO2
|
||||
* periodically through the day and overnight; the readings map to
|
||||
* `aggregation: "latest"` and sit in NEITHER `CUMULATIVE_HK_TYPES` nor
|
||||
* `HIGH_FREQUENCY_MEAN_TYPES`, so every sample piled up raw forever.
|
||||
* Folded here to hourly-mean rows with the daily min/max preserved on
|
||||
* the rollup tier (the lowest sample is the overnight-desaturation
|
||||
* nadir — clinically the meaningful figure, not the mean).
|
||||
*
|
||||
* Every member is deliberately EXEMPT from the destructive daily-mean drain
|
||||
* (`HIGH_FREQUENCY_MEAN_TYPES`); the disjointness from that set is the
|
||||
* load-bearing invariant a regression test pins — a type in both would be
|
||||
* folded by two drains at once and corrupt the value.
|
||||
*
|
||||
* `RESPIRATORY_RATE` is deliberately NOT here: it is already a member of
|
||||
* `HIGH_FREQUENCY_MEAN_TYPES`, so the nightly mean-consolidation already
|
||||
* collapses it to a daily-mean row. Adding it here would double-fold it.
|
||||
*
|
||||
* One fold grain per tier: every member folds HOURLY. A per-type grain
|
||||
* split would fork the write path and the rebuild for no reader benefit.
|
||||
*/
|
||||
export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
|
||||
new Set<MeasurementType>([
|
||||
"HEART_RATE_VARIABILITY",
|
||||
"PULSE",
|
||||
"OXYGEN_SATURATION",
|
||||
]);
|
||||
|
||||
/**
|
||||
* The sources whose raw dense samples fold, and which dense types of each.
|
||||
* A source belongs here only for a type it stores sample by sample through
|
||||
* the day; a type it stores as one value a day (a nightly SpO2 average, a
|
||||
* recovery's HRV) is no intraday stream, and folding that row would only
|
||||
* move it to another hour.
|
||||
*
|
||||
* - Apple Health and Health Connect: every dense type. Both write the
|
||||
* samples of an export raw, and without the fold every new export would
|
||||
* add another quarter of minute-by-minute heart rate.
|
||||
* - Google Health: heart rate, one row per sample from the `heart-rate`
|
||||
* collection (SpO2 comes from the daily collection). A full-history
|
||||
* backfill brought tens of millions of these onto one account (#1297).
|
||||
* - Oura: heart rate, a reading every few minutes from the `heartrate`
|
||||
* collection (SpO2 is the nightly average).
|
||||
*
|
||||
* Fitbit does not sync intraday heart rate, and WHOOP, Withings and Polar
|
||||
* write dense types only as one value per recovery, night or spot check, so
|
||||
* none of them is here.
|
||||
*
|
||||
* Each source folds into its OWN `stats:` rows, so an hour two sources both
|
||||
* covered keeps one mean per source, the same way the raw samples kept their
|
||||
* source. The walk runs in this order.
|
||||
*/
|
||||
export const DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE = {
|
||||
APPLE_HEALTH: DENSE_INTRADAY_RETENTION_TYPES,
|
||||
HEALTH_CONNECT: DENSE_INTRADAY_RETENTION_TYPES,
|
||||
GOOGLE_HEALTH: new Set<MeasurementType>(["PULSE"]),
|
||||
OURA: new Set<MeasurementType>(["PULSE"]),
|
||||
} as const satisfies Partial<
|
||||
Record<MeasurementSource, ReadonlySet<MeasurementType>>
|
||||
>;
|
||||
|
||||
export type DenseIntradayFoldSource =
|
||||
keyof typeof DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE;
|
||||
|
||||
/** Every source the fold walks, in walk order. */
|
||||
export const DENSE_INTRADAY_FOLD_SOURCES = Object.keys(
|
||||
DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
|
||||
) as readonly DenseIntradayFoldSource[];
|
||||
@@ -47,12 +47,14 @@
|
||||
* of tombstones does not use that reach; a tombstone it misses stays class B,
|
||||
* which keeps a row longer, never loses one.
|
||||
*
|
||||
* Health Connect folds its raw heart-rate history the same way under its own
|
||||
* source (the importer writes samples of the last 90 days raw, and the
|
||||
* nightly pass folds them once they age out), so the hour rule covers both
|
||||
* sources and a covering row is always looked up under the sample's own
|
||||
* source. The day rule stays Apple Health only: the nightly mean
|
||||
* consolidation reads no other source.
|
||||
* Every source the dense fold walks folds under its own source, for the
|
||||
* types it stores sample by sample (`DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE`):
|
||||
* Health Connect from its importer, and Google Health and Oura from their
|
||||
* syncs, whose backfills walk years of heart rate (#1297). So there is one
|
||||
* hour rule per source, and a covering row is always looked up under the
|
||||
* sample's own source. The provider syncs ask the same question before they
|
||||
* write (`withoutFoldedSamples`). The day rule stays Apple Health only: the
|
||||
* nightly mean consolidation reads no other source.
|
||||
*/
|
||||
import type {
|
||||
MeasurementSource,
|
||||
@@ -71,11 +73,14 @@ import {
|
||||
hourOfDayForUserTz,
|
||||
} from "@/lib/measurements/consolidation-tz";
|
||||
import {
|
||||
DENSE_INTRADAY_FOLD_SOURCES,
|
||||
DENSE_INTRADAY_RETENTION_DAYS,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
hourlyStatsExternalId,
|
||||
} from "@/lib/measurements/dense-intraday-retention";
|
||||
import {
|
||||
DENSE_INTRADAY_FOLD_SOURCES,
|
||||
DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
|
||||
DENSE_INTRADAY_RETENTION_TYPES,
|
||||
} from "@/lib/measurements/dense-intraday-sources";
|
||||
|
||||
/** The per-entry status reason the batch result carries for such a row. */
|
||||
export type FoldedWindowReason = "folded_window";
|
||||
@@ -88,33 +93,38 @@ const HOUR_MS = 3_600_000;
|
||||
* `reachMs` is how far from a sample the anchor of its window can sit in any
|
||||
* zone (see the module comment), for the timezone-change check.
|
||||
*/
|
||||
export const FOLDED_WINDOW_RULES = [
|
||||
{
|
||||
export const FOLDED_WINDOW_RULES: readonly FoldedWindowRule[] = [
|
||||
// One hour rule per source the dense fold walks, with that source's types.
|
||||
...DENSE_INTRADAY_FOLD_SOURCES.map((source): FoldedWindowRule => ({
|
||||
window: "hour",
|
||||
sources: DENSE_INTRADAY_FOLD_SOURCES,
|
||||
types: DENSE_INTRADAY_RETENTION_TYPES,
|
||||
sources: [source],
|
||||
types: DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE[source],
|
||||
foldedAfterMs: DENSE_INTRADAY_RETENTION_DAYS * 24 * HOUR_MS,
|
||||
reachMs: HOUR_MS / 2,
|
||||
},
|
||||
})),
|
||||
{
|
||||
window: "day",
|
||||
sources: ["APPLE_HEALTH"] as const,
|
||||
sources: ["APPLE_HEALTH"],
|
||||
types: HIGH_FREQUENCY_MEAN_TYPES,
|
||||
foldedAfterMs: CONSOLIDATION_GRACE_CUTOFF_HOURS * HOUR_MS,
|
||||
reachMs: 14 * HOUR_MS,
|
||||
},
|
||||
] as const;
|
||||
];
|
||||
|
||||
type FoldedWindowRule = (typeof FOLDED_WINDOW_RULES)[number];
|
||||
interface FoldedWindowRule {
|
||||
window: "hour" | "day";
|
||||
sources: readonly MeasurementSource[];
|
||||
types: ReadonlySet<MeasurementType>;
|
||||
foldedAfterMs: number;
|
||||
reachMs: number;
|
||||
}
|
||||
|
||||
function ruleFor(
|
||||
row: Pick<FoldedWindowCandidateRow, "type" | "source">,
|
||||
): FoldedWindowRule | null {
|
||||
return (
|
||||
FOLDED_WINDOW_RULES.find(
|
||||
(rule) =>
|
||||
(rule.sources as readonly MeasurementSource[]).includes(row.source) &&
|
||||
rule.types.has(row.type),
|
||||
(rule) => rule.sources.includes(row.source) && rule.types.has(row.type),
|
||||
) ?? null
|
||||
);
|
||||
}
|
||||
@@ -409,6 +419,46 @@ export async function findFoldedWindowDuplicates(
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* The rows of one provider write without the raw samples whose hour the
|
||||
* dense fold has already taken from that source (#1297).
|
||||
*
|
||||
* The provider syncs write first-write-wins on `(user, type, source,
|
||||
* externalId)`, and the fold deletes the samples it averages, so a backfill
|
||||
* or a full sync that walks old history would find no row for a folded
|
||||
* sample and insert it again beside the hourly mean that already accounts
|
||||
* for it. Google Health would even re-key the hourly row: a sample that fell
|
||||
* on the hour's anchor shares its instant, and the sync's natural-key rescue
|
||||
* then overwrote the mean with the sample. Rows too young to have been folded
|
||||
* cost no query, so the hourly syncs never ask.
|
||||
*/
|
||||
export async function withoutFoldedSamples<
|
||||
T extends { type: string; externalId?: string | null; measuredAt: Date },
|
||||
>(
|
||||
client: LookupClient,
|
||||
userId: string,
|
||||
source: MeasurementSource,
|
||||
rows: readonly T[],
|
||||
options: { now?: Date } = {},
|
||||
): Promise<{ kept: T[]; dropped: number }> {
|
||||
const folded = await findFoldedWindowDuplicates(
|
||||
client,
|
||||
userId,
|
||||
rows.map((row) => ({
|
||||
type: row.type as MeasurementType,
|
||||
source,
|
||||
externalId: row.externalId ?? null,
|
||||
measuredAt: row.measuredAt,
|
||||
})),
|
||||
options,
|
||||
);
|
||||
if (folded.size === 0) return { kept: [...rows], dropped: 0 };
|
||||
return {
|
||||
kept: rows.filter((_, index) => !folded.has(index)),
|
||||
dropped: folded.size,
|
||||
};
|
||||
}
|
||||
|
||||
/** A tombstone as the classifier reads it. */
|
||||
export interface TombstoneCandidateRow extends FoldedWindowCandidateRow {
|
||||
deletedAt: Date | null;
|
||||
|
||||
@@ -11,14 +11,16 @@
|
||||
* Writes are first-write-wins: a sample is final once it is in Oura's cloud,
|
||||
* and it is keyed on its own instant (`heartrate:<ISO>`), so a re-read of the
|
||||
* same range is an insert that `skipDuplicates` drops. A row the person
|
||||
* deleted keeps its tombstone; the re-read does not bring it back. Oura's
|
||||
* samples go raw, like Google Health's and Fitbit's minute stream: the dense
|
||||
* hourly fold covers only the export sources (`DENSE_INTRADAY_FOLD_SOURCES`).
|
||||
* deleted keeps its tombstone; the re-read does not bring it back. Samples
|
||||
* older than the retention window are folded into hourly means like Google
|
||||
* Health's (`DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE`), and a re-read of a folded
|
||||
* hour writes nothing (`withoutFoldedSamples`).
|
||||
*/
|
||||
import { prisma } from "@/lib/db";
|
||||
import type { MeasurementType } from "@/generated/prisma/client";
|
||||
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
|
||||
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
|
||||
import { withoutFoldedSamples } from "@/lib/measurements/folded-window";
|
||||
import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups";
|
||||
import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
|
||||
import { forEachHeartRatePage } from "./client";
|
||||
@@ -76,10 +78,17 @@ export async function writeOuraHeartRatePage(
|
||||
const m = mapHeartRateSample(s);
|
||||
if (m) mapped.push(m);
|
||||
}
|
||||
const kept = dropImplausibleMeasurements("oura", mapped, (m) => ({
|
||||
type: m.type,
|
||||
value: m.value,
|
||||
}));
|
||||
// A sample of an hour the dense fold already took is not written again: it
|
||||
// was deleted with the fold, so `skipDuplicates` would not stop it.
|
||||
const { kept } = await withoutFoldedSamples(
|
||||
prisma,
|
||||
userId,
|
||||
"OURA",
|
||||
dropImplausibleMeasurements("oura", mapped, (m) => ({
|
||||
type: m.type,
|
||||
value: m.value,
|
||||
})),
|
||||
);
|
||||
let inserted = 0;
|
||||
for (let i = 0; i < kept.length; i += OURA_HEART_RATE_WRITE_CHUNK) {
|
||||
const chunk = kept.slice(i, i + OURA_HEART_RATE_WRITE_CHUNK);
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
/**
|
||||
* The full backfill's windowed fold and its handler (#1298): the windows it
|
||||
* cuts, the per-statement limit every window runs under, and how a run that
|
||||
* reaches its job budget hands the remaining years on instead of failing.
|
||||
*
|
||||
* The fold itself is mocked here; that the windows write exactly the buckets
|
||||
* one pass over the whole span writes is pinned against Postgres in
|
||||
* `tests/integration/rollup-full-backfill.test.ts`.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { Job } from "pg-boss";
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
recomputeUserRollups: vi.fn(),
|
||||
bossSend: vi.fn(),
|
||||
getGlobalBoss: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({
|
||||
...(await importOriginal<
|
||||
typeof import("@/lib/rollups/measurement-rollups")
|
||||
>()),
|
||||
recomputeUserRollups: mocks.recomputeUserRollups,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: () => mocks.getGlobalBoss(),
|
||||
}));
|
||||
|
||||
import { handleRollupFullBackfill } from "@/lib/jobs/reminder/rollup-handlers";
|
||||
import {
|
||||
foldUserRollupHistory,
|
||||
fullBackfillWindows,
|
||||
ROLLUP_FULL_BACKFILL_QUEUE,
|
||||
ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
type RollupFullBackfillPayload,
|
||||
} from "@/lib/rollups/full-backfill";
|
||||
import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups";
|
||||
|
||||
const NOW = new Date("2026-10-29T10:00:00.000Z");
|
||||
|
||||
function job(
|
||||
data: Partial<RollupFullBackfillPayload> = {},
|
||||
): Job<RollupFullBackfillPayload> {
|
||||
return {
|
||||
id: "job-1",
|
||||
name: ROLLUP_FULL_BACKFILL_QUEUE,
|
||||
data: { userId: "user-1", enqueuedAt: NOW.toISOString(), ...data },
|
||||
expireInSeconds: 3600,
|
||||
signal: new AbortController().signal,
|
||||
} as unknown as Job<RollupFullBackfillPayload>;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
vi.setSystemTime(NOW);
|
||||
mocks.recomputeUserRollups.mockReset();
|
||||
mocks.recomputeUserRollups.mockResolvedValue({
|
||||
rowsUpserted: 10,
|
||||
durationMs: 1,
|
||||
});
|
||||
mocks.bossSend.mockReset();
|
||||
mocks.bossSend.mockResolvedValue("job-next");
|
||||
mocks.getGlobalBoss.mockReset();
|
||||
mocks.getGlobalBoss.mockReturnValue({ send: mocks.bossSend });
|
||||
});
|
||||
|
||||
describe("fullBackfillWindows", () => {
|
||||
it("cuts the span at UTC New Year, newest first, and covers it exactly", () => {
|
||||
const from = new Date("2021-10-30T10:00:00.000Z");
|
||||
const windows = fullBackfillWindows(from, NOW);
|
||||
expect(
|
||||
windows.map((w) => [w.from.toISOString(), w.to.toISOString()]),
|
||||
).toEqual([
|
||||
["2026-01-01T00:00:00.000Z", NOW.toISOString()],
|
||||
["2025-01-01T00:00:00.000Z", "2026-01-01T00:00:00.000Z"],
|
||||
["2024-01-01T00:00:00.000Z", "2025-01-01T00:00:00.000Z"],
|
||||
["2023-01-01T00:00:00.000Z", "2024-01-01T00:00:00.000Z"],
|
||||
["2022-01-01T00:00:00.000Z", "2023-01-01T00:00:00.000Z"],
|
||||
[from.toISOString(), "2022-01-01T00:00:00.000Z"],
|
||||
]);
|
||||
for (const w of windows) {
|
||||
expect(w.to.getTime() - w.from.getTime()).toBeLessThanOrEqual(
|
||||
366 * 86_400_000,
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
it("treats an end on New Year as the end of the year before", () => {
|
||||
const windows = fullBackfillWindows(
|
||||
new Date("2024-06-01T00:00:00.000Z"),
|
||||
new Date("2025-01-01T00:00:00.000Z"),
|
||||
);
|
||||
expect(windows).toEqual([
|
||||
{
|
||||
from: new Date("2024-06-01T00:00:00.000Z"),
|
||||
to: new Date("2025-01-01T00:00:00.000Z"),
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("is empty for an empty span", () => {
|
||||
expect(fullBackfillWindows(NOW, NOW)).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("foldUserRollupHistory", () => {
|
||||
it("folds every window of the fold window under the given statement limit", async () => {
|
||||
const done = await foldUserRollupHistory("user-1", {
|
||||
statementTimeoutMs: 1234,
|
||||
});
|
||||
const expected = fullBackfillWindows(
|
||||
new Date(NOW.getTime() - ROLLUP_FOLD_WINDOW_MS),
|
||||
NOW,
|
||||
);
|
||||
expect(mocks.recomputeUserRollups).toHaveBeenCalledTimes(expected.length);
|
||||
expected.forEach((w, i) => {
|
||||
expect(mocks.recomputeUserRollups).toHaveBeenNthCalledWith(
|
||||
i + 1,
|
||||
"user-1",
|
||||
expect.objectContaining({
|
||||
from: w.from,
|
||||
to: w.to,
|
||||
statementTimeoutMs: 1234,
|
||||
}),
|
||||
);
|
||||
});
|
||||
expect(done).toMatchObject({
|
||||
rowsUpserted: 10 * expected.length,
|
||||
windowsFolded: expected.length,
|
||||
});
|
||||
expect(done.resumeBefore).toBeUndefined();
|
||||
});
|
||||
|
||||
it("stops before the next window and names where the rest begins", async () => {
|
||||
const done = await foldUserRollupHistory("user-1", {
|
||||
shouldStop: () => mocks.recomputeUserRollups.mock.calls.length >= 1,
|
||||
});
|
||||
expect(mocks.recomputeUserRollups).toHaveBeenCalledTimes(1);
|
||||
expect(done.windowsFolded).toBe(1);
|
||||
expect(done.resumeBefore).toEqual(new Date("2026-01-01T00:00:00.000Z"));
|
||||
});
|
||||
|
||||
it("reports a window the fold stopped inside as still to fold", async () => {
|
||||
mocks.recomputeUserRollups
|
||||
.mockResolvedValueOnce({ rowsUpserted: 5, durationMs: 1 })
|
||||
.mockResolvedValueOnce({ rowsUpserted: 2, durationMs: 1, stopped: true });
|
||||
const done = await foldUserRollupHistory("user-1", {
|
||||
shouldStop: () => false,
|
||||
});
|
||||
expect(done.windowsFolded).toBe(1);
|
||||
expect(done.resumeBefore).toEqual(new Date("2026-01-01T00:00:00.000Z"));
|
||||
});
|
||||
|
||||
it("resumes with only the windows that end at or before `before`", async () => {
|
||||
await foldUserRollupHistory("user-1", {
|
||||
before: new Date("2025-01-01T00:00:00.000Z"),
|
||||
});
|
||||
const tos = mocks.recomputeUserRollups.mock.calls.map(
|
||||
(call) => (call[1] as { to: Date }).to,
|
||||
);
|
||||
expect(tos[0]).toEqual(new Date("2025-01-01T00:00:00.000Z"));
|
||||
for (const to of tos) {
|
||||
expect(to.getTime()).toBeLessThanOrEqual(
|
||||
new Date("2025-01-01T00:00:00.000Z").getTime(),
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("handleRollupFullBackfill", () => {
|
||||
it("runs every statement of the fold under the full backfill's own limit", async () => {
|
||||
await handleRollupFullBackfill([job()]);
|
||||
expect(mocks.recomputeUserRollups).toHaveBeenCalled();
|
||||
for (const call of mocks.recomputeUserRollups.mock.calls) {
|
||||
expect(call[1]).toMatchObject({
|
||||
statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
});
|
||||
}
|
||||
expect(mocks.bossSend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("hands the years it did not reach to a follow-up and completes", async () => {
|
||||
mocks.recomputeUserRollups
|
||||
.mockResolvedValueOnce({ rowsUpserted: 5, durationMs: 1 })
|
||||
.mockResolvedValueOnce({ rowsUpserted: 0, durationMs: 1, stopped: true });
|
||||
const outcome = await handleRollupFullBackfill([job()]);
|
||||
expect(outcome.ok).toBe(true);
|
||||
expect(mocks.bossSend).toHaveBeenCalledTimes(1);
|
||||
const [queue, payload, options] = mocks.bossSend.mock.calls[0];
|
||||
expect(queue).toBe(ROLLUP_FULL_BACKFILL_QUEUE);
|
||||
expect(payload).toMatchObject({
|
||||
userId: "user-1",
|
||||
resumeBefore: "2026-01-01T00:00:00.000Z",
|
||||
});
|
||||
// Not the boot key: the running job still holds that one.
|
||||
expect(options.singletonKey).toBe(
|
||||
"boot-backfill|user-1|2026-01-01T00:00:00.000Z",
|
||||
);
|
||||
expect(options.startAfter).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("resumes a follow-up where the run before it stopped", async () => {
|
||||
await handleRollupFullBackfill([
|
||||
job({ resumeBefore: "2024-01-01T00:00:00.000Z" }),
|
||||
]);
|
||||
expect(
|
||||
(mocks.recomputeUserRollups.mock.calls[0][1] as { to: Date }).to,
|
||||
).toEqual(new Date("2024-01-01T00:00:00.000Z"));
|
||||
});
|
||||
|
||||
it("fails a run that could not finish one window, so its retry starts it again", async () => {
|
||||
mocks.recomputeUserRollups.mockResolvedValueOnce({
|
||||
rowsUpserted: 0,
|
||||
durationMs: 1,
|
||||
stopped: true,
|
||||
});
|
||||
await expect(handleRollupFullBackfill([job()])).rejects.toThrow(
|
||||
/before it folded a window/,
|
||||
);
|
||||
expect(mocks.bossSend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("fails the run when the follow-up cannot be queued", async () => {
|
||||
mocks.getGlobalBoss.mockReturnValue(null);
|
||||
mocks.recomputeUserRollups
|
||||
.mockResolvedValueOnce({ rowsUpserted: 5, durationMs: 1 })
|
||||
.mockResolvedValueOnce({ rowsUpserted: 0, durationMs: 1, stopped: true });
|
||||
await expect(handleRollupFullBackfill([job()])).rejects.toThrow(/no queue/);
|
||||
});
|
||||
});
|
||||
@@ -29,6 +29,7 @@ const mocks = vi.hoisted(() => ({
|
||||
findMany: vi.fn(),
|
||||
findFirstMeasurement: vi.fn(),
|
||||
txExecuteRawUnsafe: vi.fn(),
|
||||
txExecuteRaw: vi.fn(),
|
||||
bossSend: vi.fn(),
|
||||
getGlobalBossMock: vi.fn(),
|
||||
}));
|
||||
@@ -87,10 +88,12 @@ void queryRaw;
|
||||
|
||||
import {
|
||||
ROLLUP_FULL_BACKFILL_QUEUE,
|
||||
enqueueBootTimeRollupBackfill,
|
||||
} from "../full-backfill";
|
||||
import {
|
||||
ROLLUP_RECOMPUTE_QUEUE,
|
||||
_resetEnsureUserRollupsFreshInFlightForTests,
|
||||
collapseToTypeDayKeys,
|
||||
enqueueBootTimeRollupBackfill,
|
||||
enqueueRollupRecompute,
|
||||
ensureUserRollupsFresh,
|
||||
recomputeBucketsForMeasurement,
|
||||
@@ -127,6 +130,8 @@ beforeEach(() => {
|
||||
createMany: mocks.createMany,
|
||||
},
|
||||
$executeRawUnsafe: mocks.txExecuteRawUnsafe,
|
||||
$executeRaw: mocks.txExecuteRaw,
|
||||
$queryRaw: mocks.queryRaw,
|
||||
});
|
||||
}
|
||||
return Promise.all(arg as unknown[]);
|
||||
@@ -996,10 +1001,10 @@ describe("enqueueBootTimeRollupBackfill", () => {
|
||||
|
||||
const sqlParts = queryRaw.mock.calls[0][0] as TemplateStringsArray;
|
||||
const sqlText = Array.isArray(sqlParts) ? sqlParts.join("?") : "";
|
||||
// Inner DISTINCT widens to `(user_id, type, bucket_start)`.
|
||||
expect(sqlText).toContain("date_trunc('day', m.\"measured_at\")");
|
||||
// LEFT JOIN now compares bucket_start on both sides.
|
||||
expect(sqlText).toContain('r."bucket_start" = mt."bucket_start"');
|
||||
// The day walk steps from each day's first reading to the next UTC day.
|
||||
expect(sqlText).toContain("date_trunc('day', d.at) + interval '1 day'");
|
||||
// The coverage check compares the walked day with the DAY bucket.
|
||||
expect(sqlText).toContain(`r."bucket_start" = date_trunc('day', d.at)`);
|
||||
});
|
||||
|
||||
// v1.16.10 — discovery and fold share one trailing window. Unbounded
|
||||
@@ -1017,12 +1022,14 @@ describe("enqueueBootTimeRollupBackfill", () => {
|
||||
const after = Date.now();
|
||||
const sqlParts = queryRaw.mock.calls[0][0] as TemplateStringsArray;
|
||||
const sqlText = Array.isArray(sqlParts) ? sqlParts.join("?") : "";
|
||||
// The inner scan filters on measured_at >= <fold window start>.
|
||||
// The day walk starts at measured_at >= <fold window start>.
|
||||
expect(sqlText).toContain('m."measured_at" >=');
|
||||
// The bound parameter is now − ROLLUP_FOLD_WINDOW_MS (5 years),
|
||||
// matching the `recomputeUserRollups` default the backfill worker
|
||||
// runs with.
|
||||
const boundArg = queryRaw.mock.calls[0][1] as Date;
|
||||
const boundArg = queryRaw.mock.calls[0].find(
|
||||
(arg: unknown) => arg instanceof Date,
|
||||
) as Date;
|
||||
expect(boundArg).toBeInstanceOf(Date);
|
||||
const fiveYearsMs = 5 * 365 * 24 * 60 * 60 * 1000;
|
||||
expect(boundArg.getTime()).toBeGreaterThanOrEqual(before - fiveYearsMs);
|
||||
|
||||
@@ -0,0 +1,362 @@
|
||||
/**
|
||||
* The full fold of one account's rollup tier, and the boot discovery that
|
||||
* decides which accounts need one.
|
||||
*
|
||||
* The write hooks keep the DAY bucket of each touched day current and the
|
||||
* read path warms the last 90 days, but neither folds the deep WEEK / MONTH /
|
||||
* YEAR history, and neither refills a column a migration adds to rows that
|
||||
* already exist (the accumulators of 0190, the reading extremes of 0417). The
|
||||
* full backfill does both, once per account the discovery finds.
|
||||
*
|
||||
* Sized for large accounts (#1298). The fold used to be four aggregates over
|
||||
* the whole five-year window, each held to the connection's statement limit
|
||||
* (60 s unless `DATABASE_STATEMENT_TIMEOUT_MS` says otherwise). On an account
|
||||
* with tens of millions of readings no aggregate finished in time, the job
|
||||
* failed on every boot, and the columns 0417 added never filled, so every
|
||||
* reader that needs them stayed on the raw table and timed out as well. The
|
||||
* fold now works through the window one UTC calendar year at a time, newest
|
||||
* first, and every statement of the full backfill runs under its own limit
|
||||
* (`ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS`), the shape the pulse day-mean
|
||||
* rebuild already had. Each window commits on its own, so a run that reaches
|
||||
* its job budget hands the remaining years to a follow-up job instead of
|
||||
* starting over on the retry.
|
||||
*/
|
||||
import { activeAccountSql } from "@/lib/account-deletion/active-account";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import {
|
||||
applyStatementBudget,
|
||||
recomputeUserRollups,
|
||||
ROLLUP_FOLD_WINDOW_MS,
|
||||
} from "@/lib/rollups/measurement-rollups";
|
||||
|
||||
/**
|
||||
* The boot-time full-fold queue: one job per account the discovery finds, so
|
||||
* a self-hosted instance converges on the rollup tier without an operator.
|
||||
*/
|
||||
export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill";
|
||||
|
||||
/** Boot-fold worker concurrency. Serial so we don't crowd the pool. */
|
||||
export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1;
|
||||
|
||||
/**
|
||||
* The full fold's expiry: an hour. A run stops itself at three quarters of it
|
||||
* (`jobBudget`), between two windows or two granularities, and hands what is
|
||||
* left to a follow-up, so the expiry bounds one run, not the whole account.
|
||||
* The margin above the budget covers the statement in flight, which can run
|
||||
* for `ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS`.
|
||||
*/
|
||||
export const ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS = 60 * 60;
|
||||
|
||||
/**
|
||||
* The limit of every statement the full backfill and its discovery run, in
|
||||
* place of the connection's request limit. Five minutes, as the pulse
|
||||
* day-mean rebuild uses: one calendar year of one account's readings,
|
||||
* aggregated per granularity, stays well inside it on the largest accounts
|
||||
* reported, while a statement that would run longer still ends.
|
||||
*/
|
||||
export const ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS = 5 * 60 * 1000;
|
||||
|
||||
/** Seconds a follow-up waits before it picks up the remaining years. */
|
||||
export const ROLLUP_FULL_BACKFILL_CONTINUATION_DELAY_SECONDS = 60;
|
||||
|
||||
/** Payload of one full-fold job. */
|
||||
export interface RollupFullBackfillPayload {
|
||||
userId: string;
|
||||
/** Wall-clock kick-off for debugging only. */
|
||||
enqueuedAt: string;
|
||||
/**
|
||||
* ISO instant. A follow-up folds only the windows that end at or before
|
||||
* it: the run that sent it folded everything newer.
|
||||
*/
|
||||
resumeBefore?: string;
|
||||
}
|
||||
|
||||
/** The retry policy of every full-fold send. */
|
||||
const RETRY_OPTIONS = {
|
||||
retryLimit: 3,
|
||||
retryDelay: 60,
|
||||
retryBackoff: true,
|
||||
} as const;
|
||||
|
||||
/**
|
||||
* `[from, to)` windows covering `[from, to)`, cut at UTC New Year and
|
||||
* returned newest first. Every DAY, MONTH and YEAR bucket lies inside one
|
||||
* window, and a WEEK that spans New Year is folded whole by both of its
|
||||
* windows (the fold snaps each window out to whole buckets), so the windows
|
||||
* write exactly the buckets one fold over the whole span writes. Pure.
|
||||
*/
|
||||
export function fullBackfillWindows(
|
||||
from: Date,
|
||||
to: Date,
|
||||
): Array<{ from: Date; to: Date }> {
|
||||
const out: Array<{ from: Date; to: Date }> = [];
|
||||
let end = to.getTime();
|
||||
const floor = from.getTime();
|
||||
while (end > floor) {
|
||||
const year = new Date(end).getUTCFullYear();
|
||||
const newYear = Date.UTC(year, 0, 1);
|
||||
const start = Math.max(
|
||||
floor,
|
||||
newYear === end ? Date.UTC(year - 1, 0, 1) : newYear,
|
||||
);
|
||||
out.push({ from: new Date(start), to: new Date(end) });
|
||||
end = start;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
export interface RollupHistoryFold {
|
||||
rowsUpserted: number;
|
||||
durationMs: number;
|
||||
/** Windows folded completely by this call. */
|
||||
windowsFolded: number;
|
||||
/**
|
||||
* Set when `shouldStop` ended the fold: every window that ends at or
|
||||
* before this instant is still to fold, and nothing newer is.
|
||||
*/
|
||||
resumeBefore?: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fold one account's rollup tier over the five-year window, every type and
|
||||
* granularity, one calendar year at a time, newest first. The same buckets
|
||||
* `recomputeUserRollups(userId)` writes in one pass; see
|
||||
* `fullBackfillWindows` for why the windows change nothing about them.
|
||||
*
|
||||
* `before` resumes a fold that stopped: only the windows ending at or before
|
||||
* it run. `shouldStop` is asked before each window and between the
|
||||
* granularities of a window; a window it interrupts is folded again whole by
|
||||
* the follow-up, so `resumeBefore` is always a window's end.
|
||||
*/
|
||||
export async function foldUserRollupHistory(
|
||||
userId: string,
|
||||
opts: {
|
||||
before?: Date;
|
||||
shouldStop?: () => boolean;
|
||||
statementTimeoutMs?: number;
|
||||
now?: Date;
|
||||
} = {},
|
||||
): Promise<RollupHistoryFold> {
|
||||
const startedAt = Date.now();
|
||||
const now = opts.now ?? new Date();
|
||||
const floor = new Date(now.getTime() - ROLLUP_FOLD_WINDOW_MS);
|
||||
const ceiling = opts.before && opts.before < now ? opts.before : now;
|
||||
let rowsUpserted = 0;
|
||||
let windowsFolded = 0;
|
||||
const result = (resumeBefore?: Date): RollupHistoryFold => ({
|
||||
rowsUpserted,
|
||||
durationMs: Date.now() - startedAt,
|
||||
windowsFolded,
|
||||
...(resumeBefore ? { resumeBefore } : {}),
|
||||
});
|
||||
for (const window of fullBackfillWindows(floor, ceiling)) {
|
||||
if (opts.shouldStop?.()) return result(window.to);
|
||||
const done = await recomputeUserRollups(userId, {
|
||||
from: window.from,
|
||||
to: window.to,
|
||||
statementTimeoutMs: opts.statementTimeoutMs,
|
||||
shouldStop: opts.shouldStop,
|
||||
});
|
||||
rowsUpserted += done.rowsUpserted;
|
||||
if (done.stopped) return result(window.to);
|
||||
windowsFolded += 1;
|
||||
}
|
||||
return result();
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand the windows a stopped fold did not reach to a follow-up job. Keyed on
|
||||
* the account and the resume point: the running job still holds the account's
|
||||
* boot key, and a second follow-up for the same point is the same work.
|
||||
* Throws when no queue is attached or the send fails, so the run fails and
|
||||
* its retry folds the same windows again rather than leaving them.
|
||||
*/
|
||||
export async function sendRollupFullBackfillContinuation(
|
||||
userId: string,
|
||||
resumeBefore: Date,
|
||||
): Promise<boolean> {
|
||||
const boss = getGlobalBoss();
|
||||
if (!boss) {
|
||||
throw new Error("rollup-full-backfill: no queue to hand the rest to");
|
||||
}
|
||||
const payload: RollupFullBackfillPayload = {
|
||||
userId,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
resumeBefore: resumeBefore.toISOString(),
|
||||
};
|
||||
const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, {
|
||||
...RETRY_OPTIONS,
|
||||
expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS,
|
||||
singletonKey: `boot-backfill|${userId}|${payload.resumeBefore}`,
|
||||
startAfter: ROLLUP_FULL_BACKFILL_CONTINUATION_DELAY_SECONDS,
|
||||
});
|
||||
return jobId !== null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a full fold of the account is queued or running. The dense fold
|
||||
* leaves such an account alone (`ConsolidationOptions.deferUser`): it deletes
|
||||
* the raw samples a fold still has to read, and a year the full backfill
|
||||
* reached after the dense fold would take its day figures from hourly means.
|
||||
* False when the queue schema cannot be read, as a web-only deployment has
|
||||
* no full backfill either.
|
||||
*/
|
||||
export async function isRollupFullBackfillPending(
|
||||
userId: string,
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const rows = await prisma.$queryRaw<Array<{ n: number }>>`
|
||||
SELECT count(*)::int AS n
|
||||
FROM pgboss.job
|
||||
WHERE name = ${ROLLUP_FULL_BACKFILL_QUEUE}
|
||||
AND state IN ('created', 'retry', 'active')
|
||||
AND data->>'userId' = ${userId}
|
||||
`;
|
||||
return (rows[0]?.n ?? 0) > 0;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Boot-time discovery. Finds the accounts whose rollup tier misses something
|
||||
* inside the fold window and enqueues one full fold per account.
|
||||
*
|
||||
* Two arms:
|
||||
*
|
||||
* 1. Missing coverage per day: a (user, type, UTC day) with a live reading
|
||||
* and no DAY rollup row. Per day rather than per type, because a sync
|
||||
* or a restore that writes without the write hook leaves gaps in a type
|
||||
* that already has buckets.
|
||||
* 2. A DAY rollup row inside the window whose regression accumulators
|
||||
* (0190) or reading extremes (0417) are NULL: a row folded before the
|
||||
* column existed. New folds write them, so the arm drains to empty.
|
||||
*
|
||||
* Both are bounded to the fold window. The fold writes nothing older, so an
|
||||
* older gap would flag the account on every boot and never close.
|
||||
*
|
||||
* Arm 1 used to read every live reading in the window, which on an account
|
||||
* with tens of millions of rows is the same scan that kept the fold itself
|
||||
* from finishing. It now walks the days instead: per account and type, the
|
||||
* first live reading at or after the start of each next UTC day, read off
|
||||
* the live `(user_id, type, measured_at)` index, so the cost follows the
|
||||
* number of days with readings, not the number of readings. The statement
|
||||
* also runs under the full backfill's own limit.
|
||||
*
|
||||
* Idempotent across reboots: an account drops off once it is covered, and
|
||||
* the per-account `singletonKey` coalesces a send while a fold is queued.
|
||||
* Best-effort: errors come back in the result so the worker boot never fails.
|
||||
*/
|
||||
export async function enqueueBootTimeRollupBackfill(
|
||||
// Optional boot-storm stagger, seconds. Default 0 keeps immediate semantics
|
||||
// for any non-boot caller.
|
||||
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 foldWindowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS);
|
||||
const users = await prisma.$transaction(
|
||||
async (tx) => {
|
||||
await applyStatementBudget(
|
||||
tx,
|
||||
ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
|
||||
);
|
||||
return tx.$queryRaw<Array<{ id: string }>>`
|
||||
WITH RECURSIVE pairs AS (
|
||||
SELECT u."id" AS user_id, t.type
|
||||
FROM users u
|
||||
CROSS JOIN unnest(enum_range(NULL::"measurement_type")) AS t(type)
|
||||
WHERE ${activeAccountSql('u."id"')}
|
||||
),
|
||||
days AS (
|
||||
SELECT p.user_id, p.type, f.at
|
||||
FROM pairs p
|
||||
CROSS JOIN LATERAL (
|
||||
SELECT min(m."measured_at") AS at
|
||||
FROM measurements m
|
||||
WHERE m."user_id" = p.user_id
|
||||
AND m."type" = p.type
|
||||
AND m."deleted_at" IS NULL
|
||||
AND m."measured_at" >= ${foldWindowStart}
|
||||
) f
|
||||
WHERE f.at IS NOT NULL
|
||||
UNION ALL
|
||||
SELECT d.user_id, d.type, n.at
|
||||
FROM days d
|
||||
CROSS JOIN LATERAL (
|
||||
SELECT min(m."measured_at") AS at
|
||||
FROM measurements m
|
||||
WHERE m."user_id" = d.user_id
|
||||
AND m."type" = d.type
|
||||
AND m."deleted_at" IS NULL
|
||||
AND m."measured_at" >= date_trunc('day', d.at) + interval '1 day'
|
||||
) n
|
||||
WHERE n.at IS NOT NULL
|
||||
)
|
||||
SELECT DISTINCT id FROM (
|
||||
SELECT d.user_id AS id
|
||||
FROM days d
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM measurement_rollups r
|
||||
WHERE r."user_id" = d.user_id
|
||||
AND r."type" = d.type
|
||||
AND r."granularity" = 'DAY'
|
||||
AND r."bucket_start" = date_trunc('day', d.at)
|
||||
)
|
||||
|
||||
UNION
|
||||
|
||||
SELECT r."user_id" AS id
|
||||
FROM measurement_rollups r
|
||||
WHERE r."granularity" = 'DAY'
|
||||
AND r."bucket_start" >= ${foldWindowStart}
|
||||
AND (r."sum_xy" IS NULL OR r."reading_min_value" IS NULL)
|
||||
AND ${activeAccountSql('r."user_id"')}
|
||||
) discovered
|
||||
`;
|
||||
},
|
||||
{
|
||||
maxWait: 10_000,
|
||||
timeout: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS + 30_000,
|
||||
},
|
||||
);
|
||||
|
||||
let enqueued = 0;
|
||||
let skipped = 0;
|
||||
for (const { id } of users) {
|
||||
const payload: RollupFullBackfillPayload = {
|
||||
userId: id,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
};
|
||||
const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, {
|
||||
...RETRY_OPTIONS,
|
||||
expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS,
|
||||
// Coalesce: if a backfill for this user is already queued and
|
||||
// we restart, pg-boss returns null instead of duplicating.
|
||||
singletonKey: `boot-backfill|${id}`,
|
||||
...(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),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -116,12 +116,16 @@ export async function readBestGranularityRollups(
|
||||
type: MeasurementType,
|
||||
windowDays: number,
|
||||
userPriorityJson?: unknown,
|
||||
now: Date = new Date(),
|
||||
): Promise<{
|
||||
granularity: RollupGranularity;
|
||||
rows: RollupBucketRow[];
|
||||
} | null> {
|
||||
if (!Number.isFinite(windowDays) || windowDays <= 0) return null;
|
||||
const since = new Date(Date.now() - windowDays * 24 * 60 * 60 * 1000);
|
||||
// The window trails the caller's `now`, so a read anchored on another
|
||||
// moment than the wall clock sees the same days through the rollups as
|
||||
// through the live fallback.
|
||||
const since = new Date(now.getTime() - windowDays * 24 * 60 * 60 * 1000);
|
||||
// v1.11.1 — load the source-priority blob once and thread it into every
|
||||
// granularity probe so the collapse never re-queries the user per floor.
|
||||
const priority =
|
||||
|
||||
@@ -34,9 +34,7 @@
|
||||
* took. Nothing here reads both tiers; the rollup replaces the legacy read and
|
||||
* falls back to live SQL only on a coverage miss.
|
||||
*/
|
||||
import { activeAccountSql } from "@/lib/account-deletion/active-account";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import {
|
||||
@@ -113,39 +111,6 @@ export interface RollupRecomputePayload {
|
||||
enqueuedAt: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* v1.4.35.1 — boot-time full-fold queue. The write-path hooks keep
|
||||
* the DAY bucket current per measurement, and `ensureUserRollupsFresh`
|
||||
* warm-on-read covers the trailing 90-day DAY window on cold mount,
|
||||
* but neither path folds the deep WEEK / MONTH / YEAR history for an
|
||||
* account that has measurements pre-dating the v1.4.35 deploy. This
|
||||
* queue exists so the worker boot can enqueue one full
|
||||
* `recomputeUserRollups(userId)` per uncovered user — no operator
|
||||
* action needed on self-hosted instances.
|
||||
*/
|
||||
export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill";
|
||||
|
||||
/** Boot-fold worker concurrency. Serial so we don't crowd the pool. */
|
||||
export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1;
|
||||
|
||||
/**
|
||||
* The full fold's expiry: an hour. One fold is four granularities, each an
|
||||
* aggregate held to the connection's statement timeout (60 s by default) and
|
||||
* a persist transaction of up to 120 s, so the worst case under the defaults
|
||||
* is close to thirteen minutes, too near pg-boss's fifteen-minute default to
|
||||
* rely on, and an operator who raises `DATABASE_STATEMENT_TIMEOUT_MS` (or
|
||||
* sets it to 0) lifts that bound altogether. Past the expiry pg-boss retries
|
||||
* the job, and the retry folded the same account a second time.
|
||||
*/
|
||||
export const ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS = 60 * 60;
|
||||
|
||||
/** Payload the boot-time backfill helper enqueues. */
|
||||
export interface RollupFullBackfillPayload {
|
||||
userId: string;
|
||||
/** Wall-clock kick-off for debugging only. */
|
||||
enqueuedAt: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Trailing fold window for `recomputeUserRollups` defaults AND the
|
||||
* boot-time discovery scan. The two MUST share this bound: the fold
|
||||
@@ -639,7 +604,7 @@ export async function readRollupBuckets(
|
||||
* comprehensive-aggregator + summaries-slice convention.
|
||||
*/
|
||||
/** Give the statements of a transaction their own limit (SET LOCAL). */
|
||||
async function applyStatementBudget(
|
||||
export async function applyStatementBudget(
|
||||
tx: Pick<typeof prisma, "$executeRaw">,
|
||||
ms: number,
|
||||
): Promise<void> {
|
||||
@@ -1316,165 +1281,6 @@ async function ensureUserRollupsFreshImpl(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* v1.4.35.1 — boot-time enqueue helper. Finds users with measurements
|
||||
* but zero rollup coverage and enqueues one full-fold job per account
|
||||
* onto `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 intervention.
|
||||
*
|
||||
* Idempotent across reboots:
|
||||
* - the discovery query matches accounts that either have zero
|
||||
* rollup coverage at all OR have at least one measurement type
|
||||
* whose DAY-rollup partition is empty. Once every type the user
|
||||
* has logged has at least one DAY bucket, the user drops off the
|
||||
* list and subsequent boots skip them. The v1.4.35.1 shape (zero-
|
||||
* rollups only) silently stranded users who logged a brand-new
|
||||
* type after the initial fold — their `isFullyCovered` probe stays
|
||||
* `false`, every read fans out to the live SQL aggregator, and
|
||||
* `/api/analytics` cold-mounts in tens of seconds instead of
|
||||
* 1.5-3 s. v1.4.38.5 widens discovery to per-type missing coverage.
|
||||
* - 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 enqueueBootTimeRollupBackfill(
|
||||
// Optional boot-storm stagger. When > 0 the per-user sends carry a
|
||||
// `startAfter` delay (seconds) so this full-history fold does not drain in
|
||||
// parallel with the other boot backfills onto one heavy tenant. Default 0
|
||||
// keeps immediate semantics for any non-boot caller.
|
||||
startAfterSeconds: number = 0,
|
||||
): Promise<{
|
||||
enqueued: number;
|
||||
skipped: number;
|
||||
error: string | null;
|
||||
}> {
|
||||
const boss = getGlobalBoss();
|
||||
if (!boss) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
}
|
||||
|
||||
try {
|
||||
// Users where at least one (type, day) pair in `measurements`
|
||||
// has no matching DAY rollup row.
|
||||
//
|
||||
// v1.4.39.1 — tightened from per-type to per-day. The legacy shape
|
||||
// matched users where ANY type had zero rollup rows. That stranded
|
||||
// accounts whose Withings sync / `/api/import` / admin restore
|
||||
// wrote measurements without firing the rollup write hook: as soon
|
||||
// as ONE rollup row existed for the type (even from a prior boot
|
||||
// backfill), the LEFT JOIN found a match and the user dropped off
|
||||
// the discovery list — leaving the dashboard chart's
|
||||
// `source=rollup` fast-path under-counting the missing days.
|
||||
// Switching the join to `(user, type, bucketStart)` surfaces every
|
||||
// per-day gap. Cost: the inner DISTINCT widens from
|
||||
// `(user_id, type)` (≈15 rows/user) to
|
||||
// `(user_id, type, day)` (≈5 k rows/user on a 5-year power user),
|
||||
// which is still bounded by the `(user_id, type, measured_at)`
|
||||
// index path. The LEFT JOIN walks `measurement_rollups` against
|
||||
// its composite primary key.
|
||||
//
|
||||
// v1.4.41 — the v1.4.39 W-SUM legacy-NULL `sum_value` UNION arm
|
||||
// has been retired. The maintainer's tenant converged at v1.4.40, so the
|
||||
// arm was a permanent no-op seq scan over `measurement_rollups`.
|
||||
// Per-day missing coverage remains the sole discovery anchor.
|
||||
//
|
||||
// v1.16.10 — discovery is bounded to the same trailing window the
|
||||
// fold writes (`ROLLUP_FOLD_WINDOW_MS`). Unbounded, any account
|
||||
// with measurements older than the window was re-flagged on every
|
||||
// boot: the fold can never produce buckets for those days, so the
|
||||
// per-day gap re-surfaced forever and the full backfill re-ran on
|
||||
// every restart. Days outside the window are not foldable by
|
||||
// design and must not anchor discovery.
|
||||
const foldWindowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS);
|
||||
// Two discovery arms, UNION-ed:
|
||||
//
|
||||
// 1. Per-day MISSING coverage — a (user, type, day) pair in
|
||||
// `measurements` with no matching DAY rollup row (v1.4.39.1).
|
||||
//
|
||||
// 2. v1.20.0 F6 NULL-accumulator arm — a user whose DAY rollup rows
|
||||
// predate migration 0190 (regression accumulators are NULL) needs a
|
||||
// one-time re-fold so the windowed slope / r² / sd readers compose
|
||||
// from the accumulators instead of falling back to the live REGR_*
|
||||
// scan. Anchored on `sum_xy IS NULL` (the column the read tier
|
||||
// treats as the coverage marker). New writes self-populate the
|
||||
// accumulators, so this arm drains to empty once the boot re-fold
|
||||
// catches up — mirrors the v1.4.39 sum_value rollout, which the
|
||||
// v1.4.41 note confirms converged via this same boot queue. Bounded
|
||||
// to the fold window so days outside it (un-foldable by design)
|
||||
// never re-flag a user on every boot. The same arm refills the
|
||||
// reading extremes of migration 0417 (`reading_min_value IS NULL`);
|
||||
// the extremes reader treats a NULL as uncovered until it lands.
|
||||
const users = await prisma.$queryRaw<Array<{ id: string }>>`
|
||||
SELECT DISTINCT id FROM (
|
||||
SELECT DISTINCT mt."user_id" AS id
|
||||
FROM (
|
||||
SELECT DISTINCT
|
||||
m."user_id",
|
||||
m."type",
|
||||
date_trunc('day', m."measured_at") AS bucket_start
|
||||
FROM measurements m
|
||||
WHERE m."deleted_at" IS NULL
|
||||
AND m."measured_at" >= ${foldWindowStart}
|
||||
) mt
|
||||
LEFT JOIN measurement_rollups r
|
||||
ON r."user_id" = mt."user_id"
|
||||
AND r."type" = mt."type"
|
||||
AND r."granularity" = 'DAY'
|
||||
AND r."bucket_start" = mt."bucket_start"
|
||||
WHERE r."bucket_start" IS NULL
|
||||
|
||||
UNION
|
||||
|
||||
SELECT DISTINCT r."user_id" AS id
|
||||
FROM measurement_rollups r
|
||||
WHERE r."granularity" = 'DAY'
|
||||
AND r."bucket_start" >= ${foldWindowStart}
|
||||
AND (r."sum_xy" IS NULL OR r."reading_min_value" IS NULL)
|
||||
) discovered
|
||||
WHERE ${activeAccountSql('discovered."id"')}
|
||||
`;
|
||||
|
||||
if (users.length === 0) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
}
|
||||
|
||||
let enqueued = 0;
|
||||
let skipped = 0;
|
||||
for (const { id } of users) {
|
||||
const payload: RollupFullBackfillPayload = {
|
||||
userId: id,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
};
|
||||
const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, {
|
||||
expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS,
|
||||
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: `boot-backfill|${id}`,
|
||||
...(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),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Collapse a list of `(type, day)` keys to the set of distinct
|
||||
* `(type, dayStart)` pairs. Used by the batch-ingest hooks so a 500-
|
||||
|
||||
Reference in new issue
Block a user