diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 901c7d2aa..7f8fa627f 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/src/__tests__/baselines/source-file-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 123, - "total": 145651, + "total": 145555, "entries": { "app/api/dashboard/summary/route.ts": 1032, "app/api/documents/inbound/route.ts": 811, @@ -91,7 +91,7 @@ "lib/insights/mood-status.ts": 814, "lib/insights/status-cache.ts": 812, "lib/integrations/status.ts": 1445, - "lib/jobs/apple-health-import-worker.ts": 863, + "lib/jobs/apple-health-import-worker.ts": 801, "lib/jobs/backup-restore.ts": 888, "lib/jobs/coach-nudge.ts": 1089, "lib/jobs/insight-pregenerate.ts": 1266, @@ -114,7 +114,7 @@ "lib/medications/scheduling/slot-upsert.ts": 926, "lib/notifications/senders/apns.ts": 1100, "lib/oura/client.ts": 1360, - "lib/oura/sync.ts": 917, + "lib/oura/sync.ts": 902, "lib/polar/client.ts": 1010, "lib/rollups/measurement-rollups.ts": 1496, "lib/search/search-records.ts": 996, @@ -123,7 +123,7 @@ "lib/telegram-webhook-handlers.ts": 1341, "lib/validations/inbound-documents.ts": 1130, "lib/whoop/client.ts": 917, - "lib/withings/sync-sleep.ts": 903, + "lib/withings/sync-sleep.ts": 884, "lib/withings/sync.ts": 978, "proxy.ts": 870 } diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index c20379ec3..d977290b1 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 411, - "total": 148324, + "total": 148227, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -350,7 +350,7 @@ "lib/insights/weight-status.ts :: prepareWeightStatusForUser": 598, "lib/integrations/connection-health-load.ts :: loadConnectionFacts": 227, "lib/integrations/status.ts :: recordSyncFailure": 292, - "lib/jobs/apple-health-import-worker.ts :: handleAppleHealthImport": 357, + "lib/jobs/apple-health-import-worker.ts :: handleAppleHealthImport": 300, "lib/jobs/coach-nudge.ts :: findTriggerForUser": 279, "lib/jobs/coach-nudge.ts :: runCoachNudgeTick": 292, "lib/jobs/coach-reminder-sweep.ts :: runCoachReminderSweep": 222, @@ -409,9 +409,9 @@ "lib/targets/vitals-builder.ts :: buildVitalTargets": 324, "lib/telegram-webhook-handlers.ts :: handleCallback": 599, "lib/telegram-webhook-handlers.ts :: handleTextMessage": 337, - "lib/withings/sync-activity.ts :: syncUserActivity": 391, - "lib/withings/sync-ecg.ts :: syncUserEcg": 213, - "lib/withings/sync-sleep.ts :: syncUserSleep": 429, + "lib/withings/sync-activity.ts :: syncUserActivity": 377, + "lib/withings/sync-ecg.ts :: syncUserEcg": 202, + "lib/withings/sync-sleep.ts :: syncUserSleep": 414, "lib/withings/sync.ts :: syncUserMeasurements": 373, "proxy.ts :: proxy": 499 } diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index aa461064a..df40d555c 100644 --- a/src/__tests__/baselines/version-stamp-comments.json +++ b/src/__tests__/baselines/version-stamp-comments.json @@ -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": 2959, - "total": 10520, + "total": 10517, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -2346,7 +2346,7 @@ "lib/jobs/__tests__/whoop-resource-sync.test.ts": 1, "lib/jobs/__tests__/withings-queues.test.ts": 2, "lib/jobs/__tests__/workout-insight-queue.test.ts": 1, - "lib/jobs/apple-health-import-worker.ts": 5, + "lib/jobs/apple-health-import-worker.ts": 4, "lib/jobs/coach-message-cleanup.ts": 1, "lib/jobs/coach-nudge-ai.ts": 3, "lib/jobs/coach-nudge.ts": 17, @@ -2953,9 +2953,9 @@ "lib/withings/client.ts": 5, "lib/withings/oauth-state.ts": 1, "lib/withings/response-classifier.ts": 1, - "lib/withings/sync-activity.ts": 8, + "lib/withings/sync-activity.ts": 7, "lib/withings/sync-ecg.ts": 4, - "lib/withings/sync-sleep.ts": 10, + "lib/withings/sync-sleep.ts": 9, "lib/withings/sync.ts": 9, "lib/withings/webhook-handler.ts": 3, "lib/workouts/list-read.ts": 1, diff --git a/src/lib/cache/closed-block.ts b/src/lib/cache/closed-block.ts index 6a4728e94..fa1408306 100644 --- a/src/lib/cache/closed-block.ts +++ b/src/lib/cache/closed-block.ts @@ -21,3 +21,20 @@ export function closedBlockUntouched(key: string, touchedFrom: Date): boolean { const match = CLOSED_BLOCK_RE.exec(key); return match !== null && Number(match[1]) <= touchedFrom.getTime(); } + +/** + * The earliest `measuredAt` among rows a write touched: the `touchedFrom` a + * writer that knows every row it wrote hands to `invalidateUserMeasurements`. + * `undefined` for no rows, which makes the invalidation drop closed blocks too. + */ +export function earliestMeasuredAt( + rows: Iterable<{ measuredAt: Date }>, +): Date | undefined { + let earliest: Date | undefined; + for (const row of rows) { + if (earliest === undefined || row.measuredAt < earliest) { + earliest = row.measuredAt; + } + } + return earliest; +} diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 1023cc0f8..f2cbfe0d9 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -26,7 +26,7 @@ */ import { unlinkSync } from "node:fs"; import { getHeapStatistics } from "node:v8"; -import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import type { PrismaClient } from "@/generated/prisma/client"; import { prisma, toJson } from "@/lib/db"; import type { Job } from "pg-boss"; @@ -40,18 +40,13 @@ import { type WorkoutRouteReference, } from "@/lib/apple-health/route-import"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation"; -import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { afterBulkImport } from "@/lib/jobs/import-tail"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { streamParseExportXml, type ImportJobProgress, type ImportJobResult, } from "@/lib/measurements/import-apple-health-export"; -import { - recomputeUserRollups, - ROLLUP_FOLD_WINDOW_MS, -} from "@/lib/rollups/measurement-rollups"; import { withBackgroundEvent } from "@/lib/logging/background"; import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; import { @@ -589,70 +584,13 @@ export async function handleAppleHealthImport( }, }); - // v1.5.0 — fold the persistent rollup table for the imported - // user. The per-row write hooks are intentionally skipped on the - // streaming-ingest path (a 100k-row import would otherwise pay - // 100k DAY-recompute round-trips); we run the rollup once at - // the end, scoped to the user's full measurement span, so - // post-import reads of the analytics + comprehensive surfaces - // hit the warm rollup table on first paint. - // - // The miss rides out as a fact rather than as a failed job: the - // import is committed and its mirror row already reads `done`, so - // failing here would report a completed import as failed and - // contradict the row the status poll surfaces to the operator. - let rollupFailed = false; - try { - // Folded over what this import wrote, not over the account's whole - // history, and never further back than the fold window every other - // path keeps to: buckets older than `ROLLUP_FOLD_WINDOW_MS` are by - // design never written, so reads there fall back to live SQL. The old - // span ran from the account's first measurement, whatever the import - // carried, and folded any history older than the window as well. - const span = result.measuredSpan; - if (span) { - const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); - const spanFrom = new Date(span.from); - const from = spanFrom > windowStart ? spanFrom : windowStart; - // `to` is exclusive in the aggregator, so one millisecond past the - // newest row. - const to = new Date(new Date(span.to).getTime() + 1); - if (to > from) { - await recomputeUserRollups(userId, { from, to }); - } - } - } catch (rollupErr) { - // Rollup failure is non-fatal — the next read falls through to - // live aggregation. Log but don't poison the import. - rollupFailed = true; - console.warn( - `[apple-health-import] Rollup recompute failed for user ${userId}`, - rollupErr, - ); - } - - // The imported types' cached status assessments describe data that just - // changed; re-warm them, the second leg every other measurement write - // runs (`afterMeasurementMutation`). The rollups above are refolded over - // the import's span rather than per day, so the leg is called directly. - const importedTypes = Object.entries(result.perType) - .filter(([, stat]) => stat.inserted + stat.updated > 0) - .map(([type]) => type as MeasurementType); - if (importedTypes.length > 0) { - // Registered so the test suite can await the detached handle. - trackBackgroundTask( - invalidateStatusInsightsForTypes(userId, importedTypes, { - // An export file the person chose to import, narrated like any - // other import rather than held to the sync gap. - source: "manual", - }).catch((err) => { - console.warn( - `[apple-health-import] status-insight invalidate failed for user ${userId}`, - err, - ); - }), - ); - } + // Rollups, measurement caches and status assessments over what the + // import wrote (`afterBulkImport`). + const { rollupFailed } = await afterBulkImport( + userId, + result, + "apple-health-import", + ); // Best-effort cleanup. A failed unlink is not fatal — `/tmp` is // periodically swept on the host. diff --git a/src/lib/jobs/health-connect-import-worker.ts b/src/lib/jobs/health-connect-import-worker.ts index b4928e022..08850c72a 100644 --- a/src/lib/jobs/health-connect-import-worker.ts +++ b/src/lib/jobs/health-connect-import-worker.ts @@ -19,7 +19,6 @@ import type { Job } from "pg-boss"; import { caughtAs, logCaught } from "@/lib/logging/signal"; -import type { MeasurementType } from "@/generated/prisma/client"; import { prisma, toJson } from "@/lib/db"; import { extractHealthConnectDb } from "@/lib/import/unzip-export-xml"; import { @@ -28,15 +27,10 @@ import { } from "@/lib/import/health-connect/import"; import { HealthConnectImportError } from "@/lib/import/health-connect/reader"; import { connectedDirectIntegrations } from "@/lib/import/health-connect/skip-packages"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; -import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { afterBulkImport } from "@/lib/jobs/import-tail"; import { annotate } from "@/lib/logging/context"; import { withBackgroundEvent } from "@/lib/logging/background"; -import { - recomputeUserRollups, - ROLLUP_FOLD_WINDOW_MS, -} from "@/lib/rollups/measurement-rollups"; import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; export const HEALTH_CONNECT_IMPORT_QUEUE = "health-connect-import"; @@ -203,43 +197,13 @@ export async function runHealthConnectImport( }, }); - // The per-row tails are skipped on a bulk import; refold the rollups over - // what this import wrote, inside the window every other path keeps to, - // and re-warm the cached assessments of the types it touched. - let rollupFailed = false; - const span = result.measuredSpan; - if (span) { - try { - const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); - const spanFrom = new Date(span.from); - const from = spanFrom > windowStart ? spanFrom : windowStart; - const to = new Date(new Date(span.to).getTime() + 1); - if (to > from) await recomputeUserRollups(userId, { from, to }); - } catch (err) { - rollupFailed = true; - console.warn( - `[health-connect-import] Rollup recompute failed for user ${userId}`, - err, - ); - } - } - const touched = Object.entries(result.perType) - .filter(([, stat]) => stat.inserted + stat.updated > 0) - .map(([type]) => type as MeasurementType); - if (touched.length > 0) { - trackBackgroundTask( - invalidateStatusInsightsForTypes(userId, touched, { - // An export file the person chose to import, like the Apple Health - // import: narrated like a hand entry, not held to the sync gap. - source: "manual", - }).catch((err) => { - console.warn( - `[health-connect-import] status-insight invalidate failed for user ${userId}`, - err, - ); - }), - ); - } + // Rollups, measurement caches and status assessments over what the + // import wrote, as the Apple Health import does (`afterBulkImport`). + const { rollupFailed } = await afterBulkImport( + userId, + result, + "health-connect-import", + ); annotate({ action: { name: "import.health-connect.done" }, diff --git a/src/lib/jobs/import-tail.ts b/src/lib/jobs/import-tail.ts new file mode 100644 index 000000000..a63dbf71d --- /dev/null +++ b/src/lib/jobs/import-tail.ts @@ -0,0 +1,82 @@ +/** + * The tail of a bulk export import (Apple Health, Health Connect), run once + * the import is committed. The per-row write hooks are skipped while the + * export streams in (a 100k-row import would otherwise pay 100k DAY + * recomputes), so the three legs every other measurement write runs through + * `afterMeasurementMutation` run here once, over what the import wrote: + * + * 1. the rollup table, refolded over the span of the rows written and never + * further back than the fold window every other path keeps to (buckets + * older than `ROLLUP_FOLD_WINDOW_MS` are by design never written, so + * reads there fall back to live SQL); + * 2. the measurement read caches, marked stale back to the oldest sample + * written. An export carries years of history, and a closed block that a + * later sync leaves in place would otherwise keep serving the pre-import + * all-time values. An import that wrote no measurement (only workouts or + * ECGs) drops them all; + * 3. the cached status assessments of the types written, narrated like a + * hand entry: an export is a file the person chose to import. + * + * A rollup miss rides out as a fact rather than a failed job: the import is + * committed and its row already reads `done`, so failing here would report a + * completed import as failed. + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation"; +import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { + recomputeUserRollups, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; + +export async function afterBulkImport( + userId: string, + result: { + measuredSpan: { from: string; to: string } | null; + perType: Record; + }, + label: string, +): Promise<{ rollupFailed: boolean }> { + const span = result.measuredSpan; + let rollupFailed = false; + if (span) { + try { + const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + const spanFrom = new Date(span.from); + const from = spanFrom > windowStart ? spanFrom : windowStart; + // `to` is exclusive in the aggregator: one millisecond past the newest. + const to = new Date(new Date(span.to).getTime() + 1); + if (to > from) await recomputeUserRollups(userId, { from, to }); + } catch (err) { + rollupFailed = true; + console.warn( + `[${label}] Rollup recompute failed for user ${userId}`, + err, + ); + } + } + + invalidateUserMeasurements( + userId, + span ? { touchedFrom: new Date(span.from) } : undefined, + ); + + const types = Object.entries(result.perType) + .filter(([, stat]) => stat.inserted + stat.updated > 0) + .map(([type]) => type as MeasurementType); + if (types.length > 0) { + // Registered so the test suite can await the detached handle. + trackBackgroundTask( + invalidateStatusInsightsForTypes(userId, types, { + source: "manual", + }).catch((err) => { + console.warn( + `[${label}] status-insight invalidate failed for user ${userId}`, + err, + ); + }), + ); + } + return { rollupFailed }; +} diff --git a/src/lib/oura/sync-heart-rate.ts b/src/lib/oura/sync-heart-rate.ts index 17597a23e..48f6d66b3 100644 --- a/src/lib/oura/sync-heart-rate.ts +++ b/src/lib/oura/sync-heart-rate.ts @@ -19,12 +19,8 @@ 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 { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; -import { getEvent } from "@/lib/logging/context"; +import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import { forEachHeartRatePage, mapHeartRateSample, @@ -120,23 +116,13 @@ export async function refoldOuraTouchedDays( touched: TouchedDays, ): Promise { if (touched.size === 0) return; - const keys = touched.list(); - try { - await recomputeBucketsForKeys(userId, keys); - invalidateStatusInsightsForTypes( - userId, - [...new Set(keys.map((k) => k.type))], - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `oura: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `oura: heart-rate rollup recompute failed for ${userId}: ${err}`, - ); - } + // The shared tail: the rollups, the measurement read caches back to the + // first touched day (a backfill window reaches years back), and the cached + // assessments of the types. + await afterMeasurementMutation(userId, touched.list(), { + source: "sync", + label: "oura heart rate", + }); } /** diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index c205e05e5..fe4f89984 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -45,17 +45,12 @@ import { toFailureKind, type FailureKind, } from "@/lib/integrations/status"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; import { emitInsertedMeasurementArrivals, type InsertedMeasurementArrivalRow, } from "@/lib/arrivals/measurement-emit"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import type { SyncWriteResult } from "@/lib/outcome/written-outcome"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; import { @@ -895,23 +890,13 @@ export async function upsertOuraMeasurements( () => {}, ); } - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `oura: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `oura: rollup recompute failed for ${userId}: ${err}`, - ); - } + // The shared tail: the rollups, the measurement read caches back to the + // oldest reading written (a history backfill writes through here and has + // no other invalidation), and the cached assessments of the types. + await afterMeasurementMutation(userId, touched, { + source: "sync", + label: "oura", + }); return imported; } diff --git a/src/lib/rollups/after-measurement-mutation.ts b/src/lib/rollups/after-measurement-mutation.ts index aa501d1d6..c96bbecdf 100644 --- a/src/lib/rollups/after-measurement-mutation.ts +++ b/src/lib/rollups/after-measurement-mutation.ts @@ -12,6 +12,16 @@ * inherently: a day left empty drops its rollup row. * 2. the cached per-metric status assessments — re-warmed via * `invalidateStatusInsightsForTypes` (fire-and-forget). + * 3. the measurement read caches (analytics, targets, derived, workouts), + * marked stale back to the earliest touched reading once the rollups + * are refolded. Without it a writer that relied on the tail alone (the + * CSV and JSON imports, the score engines, the WHOOP webhook, the Withings + * activity, sleep and ECG legs, the Oura history backfill) left + * every cached read, the closed history blocks above all, on the + * pre-write values until their TTL. A caller that invalidated already + * (an interactive write evicts) loses nothing: marking stale after an + * eviction only catches an entry rebuilt from the old rollups between + * the two. * * Both legs are best-effort: a populator hiccup never fails the user's * mutation. Callers collect the affected `(type, measuredAt)` pairs @@ -27,6 +37,8 @@ import { type StatusIngestSource, } from "@/lib/insights/status-invalidation"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { earliestMeasuredAt } from "@/lib/cache/closed-block"; import type { MeasurementType } from "@/generated/prisma/client"; export async function afterMeasurementMutation( @@ -54,6 +66,7 @@ export async function afterMeasurementMutation( } catch (err) { console.warn(`[${label}] rollup recompute failed`, err); } + invalidateUserMeasurements(userId, { touchedFrom: earliestMeasuredAt(rows) }); // Status-insight re-warm is fire-and-forget — it must run even when // the rollup leg above failed, so the two legs never gate each other. diff --git a/src/lib/sleep/sweep-stale-segments.ts b/src/lib/sleep/sweep-stale-segments.ts index e1503d692..b5607c1d4 100644 --- a/src/lib/sleep/sweep-stale-segments.ts +++ b/src/lib/sleep/sweep-stale-segments.ts @@ -30,6 +30,7 @@ import type { MeasurementSource } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { getEvent } from "@/lib/logging/context"; /** One just-fetched sleep session: its externalId prefix + the fresh ids. */ @@ -79,5 +80,9 @@ export async function sweepStaleSleepSegments( ); } } + // A tombstoned segment changes the nights the cached reads hold. The sweep + // does not know how far back the swept rows reach, so every cached block + // goes; it removes rows only when a session was re-scored, which is rare. + if (removed > 0) invalidateUserMeasurements(userId); return removed; } diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index 9fa6732c5..b5199cb0a 100644 --- a/src/lib/withings/sync-activity.ts +++ b/src/lib/withings/sync-activity.ts @@ -44,11 +44,7 @@ import { getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; import { getUnitForType } from "@/lib/validations/measurement"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import { hasActivityScope, @@ -578,26 +574,12 @@ export async function syncUserActivity( // v1.4.39.1 — refresh the persistent rollup table for every distinct // (type, day) the sync touched. Collapsed so a multi-day Withings // catch-up costs at most ~N (type, day) recomputes. - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - - // v1.8.0 — drop the per-metric assessment caches the synced types - // dirty (steps feed the general overview). Fire-and-forget. - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `withings activity: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `withings activity: rollup recompute failed for ${userId}: ${err}`, - ); - } + // The shared tail: the rollups, the measurement read caches back to the + // oldest touched reading, and the cached assessments of the types. + await afterMeasurementMutation(userId, touched, { + source: "sync", + label: "withings activity", + }); // Only a clean write stamps success. A swallowed batch-write left the // ledger honest via the transient failure recorded above. diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts index 865532dc5..477a8de07 100644 --- a/src/lib/withings/sync-ecg.ts +++ b/src/lib/withings/sync-ecg.ts @@ -54,11 +54,7 @@ import { prisma } from "@/lib/db"; import { annotate, getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; import { persistEcgRecording } from "@/lib/ecg/persist-recording"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import { WithingsApiError, @@ -517,23 +513,12 @@ export async function syncUserEcg( // Refresh the persistent rollup tier for every distinct (type, day) touched, // then drop the per-metric assessment caches the new events dirty. - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `withings ecg: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `withings ecg: rollup recompute failed for ${userId}: ${err}`, - ); - } + // The shared tail: the rollups, the measurement read caches back to the + // oldest touched reading, and the cached assessments of the types. + await afterMeasurementMutation(userId, touched, { + source: "sync", + label: "withings ecg", + }); if (sourceWriteError) { await recordWithingsSyncFailure( diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index 25b7abaca..c8b99eaf6 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/src/lib/withings/sync-sleep.ts @@ -43,11 +43,7 @@ import { prisma } from "@/lib/db"; import { annotate, getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; import { getUnitForType } from "@/lib/validations/measurement"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; import { @@ -693,27 +689,12 @@ export async function syncUserSleep( // v1.4.39.1 — refresh the persistent rollup table for every distinct // (type, day) the sync touched. Sleep segments collapse heavily — // ~10 stage rows per night land in one DAY recompute. - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - - // v1.8.0 — drop the per-metric assessment caches the synced types - // dirty (sleep + resting-HR feed the pulse / general cards). - // Fire-and-forget. - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `withings sleep: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `withings sleep: rollup recompute failed for ${userId}: ${err}`, - ); - } + // The shared tail: the rollups, the measurement read caches back to the + // oldest touched reading, and the cached assessments of the types. + await afterMeasurementMutation(userId, touched, { + source: "sync", + label: "withings sleep", + }); // If a stage segment for last night just landed, kick the debounced // morning refresh so the digest/score finalise with the current sleep. The diff --git a/tests/integration/bulk-writer-cache-invalidation.test.ts b/tests/integration/bulk-writer-cache-invalidation.test.ts new file mode 100644 index 000000000..81ce584cc --- /dev/null +++ b/tests/integration/bulk-writer-cache-invalidation.test.ts @@ -0,0 +1,176 @@ +/** + * Every bulk writer of measurements marks the cached reads stale back to the + * oldest reading it wrote. + * + * A closed analytics block (a value over readings before a fixed instant) + * survives every write that says it reached no further back, so a writer that + * did not invalidate at all left the block on its pre-write value: an Apple + * Health export carrying years of history showed the old all-time figures + * until the entry aged out, and the next sync, reaching back only a day, + * spared it. Pinned here for the writers that had no invalidation: the Apple + * Health import worker, the shared post-mutation tail the CSV and JSON + * imports and the score engines rely on, the Oura upsert a history backfill + * writes through, and the sleep-segment sweep. + */ +import { createHash } from "node:crypto"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { strToU8, zipSync } from "fflate"; +import type { Job } from "pg-boss"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { closedBlockKeyPart } from "@/lib/cache/closed-block"; +import { caches, type ServerCache } from "@/lib/cache/server-cache"; +import { + APPLE_HEALTH_IMPORT_PARSER_REVISION, + handleAppleHealthImport, + _setWorkerPrismaForTests, + type AppleHealthImportPayload, +} from "@/lib/jobs/apple-health-import-worker"; +import { upsertOuraMeasurements } from "@/lib/oura/sync"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import { sweepStaleSleepSegments } from "@/lib/sleep/sweep-stale-segments"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +const prisma = getPrismaClient(); +const analytics = caches.analytics as ServerCache<{ v: number }>; +const USER = "user-bulk-writer-cache"; +const DAY_MS = 86_400_000; +const NOW = Date.now(); +const yearsAgo = (years: number) => new Date(NOW - years * 365 * DAY_MS); + +/** A closed block over the readings before `before`, and an open entry. */ +function seedCache(before: Date) { + const closed = `${USER}|pre-fold-remainder|${closedBlockKeyPart(before)}|x`; + const open = `${USER}|dashboard-snapshot|en`; + analytics.deleteByPrefix(`${USER}|`); + analytics.set(closed, { v: 1 }); + analytics.set(open, { v: 2 }); + return { closed, open }; +} + +beforeEach(async () => { + await truncateAllTables(prisma); + _setWorkerPrismaForTests(prisma); + await prisma.user.create({ + data: { + id: USER, + username: USER, + email: `${USER}@example.test`, + timezone: "UTC", + }, + }); +}); + +function record(at: Date, value: number): string { + const stamp = at.toISOString().replace("T", " ").slice(0, 19) + " +0000"; + return ``; +} + +async function runImport(xmlRecords: string[]) { + const xml = ` + + ${xmlRecords.join("\n ")} +`; + const dir = mkdtempSync(join(tmpdir(), "hl-import-cache-")); + const path = join(dir, "export.zip"); + const bytes = Buffer.from( + zipSync({ "apple_health_export/export.xml": strToU8(xml) }), + ); + writeFileSync(path, bytes); + await prisma.importJob.create({ + data: { + userId: USER, + pgBossJobId: "cache-job", + status: "queued", + uploadBytes: bytes.length, + uploadSha256: createHash("sha256").update(bytes).digest("hex"), + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + }, + }); + const outcome = await handleAppleHealthImport({ + id: "cache-job", + data: { + userId: USER, + uploadPath: path, + uploadBytes: bytes.length, + enqueuedAt: new Date().toISOString(), + }, + } as unknown as Job); + expect(outcome.ok).toBe(true); +} + +describe("bulk writers invalidate the measurement read caches", () => { + it("the Apple Health import marks stale every block its oldest sample reaches", async () => { + const reached = seedCache(yearsAgo(1)); + const older = `${USER}|pre-fold-remainder|${closedBlockKeyPart(yearsAgo(9))}|x`; + analytics.set(older, { v: 3 }); + + await runImport([record(yearsAgo(7), 88), record(yearsAgo(0.1), 84)]); + + expect(analytics.get(reached.closed)).toBeNull(); + expect(analytics.get(reached.open)).toBeNull(); + // A block of readings older than anything imported stays. + expect(analytics.get(older)).toEqual({ v: 3 }); + }); + + it("the shared post-mutation tail invalidates back to its oldest row", async () => { + const cache = seedCache(yearsAgo(1)); + await afterMeasurementMutation( + USER, + [{ type: "WEIGHT", measuredAt: yearsAgo(3) }], + { source: "manual" }, + ); + expect(analytics.get(cache.closed)).toBeNull(); + expect(analytics.get(cache.open)).toBeNull(); + }); + + it("the Oura upsert a history backfill writes through invalidates", async () => { + const cache = seedCache(yearsAgo(1)); + const written = await upsertOuraMeasurements( + USER, + [ + { + type: "RESTING_HEART_RATE", + value: 52, + unit: "bpm", + measuredAt: yearsAgo(2), + externalId: "oura:test:resting:1", + }, + ], + { emitArrivals: false }, + ); + expect(written).toBe(1); + expect(analytics.get(cache.closed)).toBeNull(); + }); + + it("a sleep-segment sweep that tombstones a row invalidates", async () => { + await prisma.measurement.create({ + data: { + userId: USER, + type: "SLEEP_DURATION", + value: 60, + unit: "min", + source: "OURA", + measuredAt: yearsAgo(2), + externalId: "night-1:stale", + }, + }); + const cache = seedCache(yearsAgo(1)); + const removed = await sweepStaleSleepSegments(USER, "OURA", [ + { prefix: "night-1:", keepIds: ["night-1:fresh"] }, + ]); + expect(removed).toBe(1); + expect(analytics.get(cache.closed)).toBeNull(); + }); +});