From e626f2123ff922bd7bc3844c47a785a490435b9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 23 Oct 2026 13:12:23 +0200 Subject: [PATCH] refactor(integrations): run one write step after every provider sync `integrations/sync-kit/after-write.ts` is the tail every provider write ends in: the arrivals go out, the touched days are refolded, the read caches are marked stale back to the earliest touched reading, and the touched status cards are re-warmed (through `afterMeasurementMutation`, shared with the interactive routes); Withings also evicts the dashboard snapshot. A full-history backfill refolds its whole touched range once through `afterProviderBackfill`. Withings, WHOOP, Polar, Nightscout, Google Health, Fitbit and Oura drop their hand-written tails, and no provider module imports the comprehensive insight generator any more; a guard holds the provider folders to that. Differences from the copies: the status cards are re-warmed even when the fold failed, and the caches are marked stale back to the earliest touched reading instead of wholesale, so older closed blocks survive a sync. Several provider unit tests mocked an export the old tails never reached (their failures were caught and only logged); the mocks now name what the code calls. --- .../baselines/source-file-lines.json | 10 +- .../baselines/source-function-lines.json | 11 +- .../baselines/version-stamp-comments.json | 4 +- .../measurement-write-tail-guard.test.ts | 33 +++-- .../provider-ai-boundary-guard.test.ts | 70 +++++++++ src/lib/fitbit/__tests__/poll-cohort.test.ts | 2 +- .../fitbit/__tests__/sync-dead-token.test.ts | 2 +- .../__tests__/sync-plausibility.test.ts | 2 +- .../__tests__/sync-sleep-replace.test.ts | 2 +- .../__tests__/sync-upsert-resurrect.test.ts | 2 +- .../fitbit/__tests__/sync-user-fitbit.test.ts | 2 +- src/lib/fitbit/__tests__/sync.test.ts | 3 +- src/lib/fitbit/sync-core.ts | 27 +--- src/lib/fitbit/sync.ts | 23 +-- .../sync-deferred-rollup-ledger.test.ts | 2 +- .../__tests__/sync-failsoft.test.ts | 2 +- .../__tests__/sync-fairness.test.ts | 2 +- .../sync-intraday-daily-wide.test.ts | 2 +- .../__tests__/sync-outcomes.test.ts | 2 +- .../__tests__/sync-plausibility.test.ts | 2 +- .../__tests__/sync-sleep-replace.test.ts | 2 +- .../__tests__/sync-time-budget.test.ts | 2 +- .../__tests__/sync-upsert-resurrect.test.ts | 2 +- src/lib/google-health/sync-core.ts | 33 ++--- src/lib/google-health/sync.ts | 24 +-- src/lib/integrations/sync-kit/after-write.ts | 140 ++++++++++++++++++ .../__tests__/fitbit-sleep-repair.test.ts | 2 +- .../google-health-sleep-repair.test.ts | 2 +- .../__tests__/sync-plausibility.test.ts | 3 +- src/lib/nightscout/__tests__/sync.test.ts | 2 +- src/lib/nightscout/sync.ts | 38 +---- src/lib/oura/__tests__/sync.test.ts | 2 +- .../__tests__/workouts-heart-rate.test.ts | 2 +- src/lib/oura/sync.ts | 28 ++-- src/lib/polar/__tests__/sync.test.ts | 3 +- src/lib/polar/sync.ts | 37 +---- src/lib/rollups/after-measurement-mutation.ts | 14 +- .../whoop/__tests__/sync-dead-token.test.ts | 2 +- .../whoop/__tests__/sync-user-whoop.test.ts | 2 +- src/lib/whoop/__tests__/sync.test.ts | 2 +- src/lib/whoop/sync-core.ts | 45 ++---- .../__tests__/subscription-state.test.ts | 2 +- src/lib/withings/__tests__/sync-ecg.test.ts | 4 - .../__tests__/sync-measure-watermark.test.ts | 13 +- src/lib/withings/__tests__/sync-plan.test.ts | 2 +- .../__tests__/sync-plausibility.test.ts | 2 +- src/lib/withings/sync.ts | 67 ++------- .../integration/provider-after-write.test.ts | 99 +++++++++++++ 48 files changed, 474 insertions(+), 307 deletions(-) create mode 100644 src/__tests__/provider-ai-boundary-guard.test.ts create mode 100644 src/lib/integrations/sync-kit/after-write.ts create mode 100644 tests/integration/provider-after-write.test.ts diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 4b039dc0a..728d41357 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": 116, - "total": 136645, + "total": 136568, "entries": { "app/api/dashboard/summary/route.ts": 1032, "app/api/documents/inbound/route.ts": 811, @@ -70,12 +70,12 @@ "lib/export/coach-backup.ts": 1070, "lib/export/profile-backup.ts": 1019, "lib/fitbit/client.ts": 1262, - "lib/fitbit/sync-core.ts": 968, + "lib/fitbit/sync-core.ts": 951, "lib/gamification/achievements-result.ts": 1094, "lib/gamification/achievements.ts": 913, "lib/google-health/client.ts": 1589, "lib/google-health/mappers.ts": 1295, - "lib/google-health/sync-core.ts": 1634, + "lib/google-health/sync-core.ts": 1621, "lib/idempotency.ts": 879, "lib/illness/correlation.ts": 1059, "lib/import/health-connect/import.ts": 1689, @@ -108,7 +108,7 @@ "lib/medications/scheduling/slot-upsert.ts": 926, "lib/notifications/senders/apns.ts": 1100, "lib/oura/client.ts": 1089, - "lib/oura/sync.ts": 902, + "lib/oura/sync.ts": 900, "lib/polar/client.ts": 893, "lib/rollups/measurement-rollups.ts": 1496, "lib/search/search-records.ts": 996, @@ -117,7 +117,7 @@ "lib/telegram-webhook-handlers.ts": 1341, "lib/validations/inbound-documents.ts": 1130, "lib/withings/sync-sleep.ts": 884, - "lib/withings/sync.ts": 978, + "lib/withings/sync.ts": 933, "proxy.ts": 870 } } diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index 7dbef3259..d7f5955f3 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": 403, - "total": 144691, + "count": 402, + "total": 144428, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -324,12 +324,11 @@ "lib/export/vaccinations-backup.ts :: restoreVaccinationsData": 216, "lib/export/visits-backup.ts :: restoreVisitsData": 261, "lib/fhir/build-bundle.ts :: buildFhirDocumentBundle": 363, - "lib/fitbit/sync-core.ts :: upsertFitbitMeasurements": 370, + "lib/fitbit/sync-core.ts :: upsertFitbitMeasurements": 356, "lib/gamification/achievements-result.ts :: buildAchievementsResult": 586, "lib/google-health/client.ts :: forEachDataPointPage": 288, "lib/google-health/client.ts :: forEachDataPointPage > walk": 223, - "lib/google-health/sync-core.ts :: upsertGoogleHealthMeasurements": 463, - "lib/google-health/sync.ts :: runGoogleHealthSyncCycle": 203, + "lib/google-health/sync-core.ts :: upsertGoogleHealthMeasurements": 453, "lib/illness/correlation.ts :: computeIllnessCorrelation": 233, "lib/import/labs-csv.ts :: importLabsCsv": 309, "lib/import/labs-csv.ts :: importLabsCsv > flush": 231, @@ -404,7 +403,7 @@ "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, + "lib/withings/sync.ts :: syncUserMeasurements": 337, "proxy.ts :: proxy": 499 } } diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index 7f96efd81..c65b301db 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": 2921, - "total": 10172, + "total": 10169, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -2918,7 +2918,7 @@ "lib/withings/sync-activity.ts": 7, "lib/withings/sync-ecg.ts": 4, "lib/withings/sync-sleep.ts": 9, - "lib/withings/sync.ts": 9, + "lib/withings/sync.ts": 6, "lib/withings/webhook-handler.ts": 3, "lib/workouts/list-read.ts": 1, "proxy.ts": 29, diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 1faa22114..471e3299d 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -32,6 +32,13 @@ const WRITE_RE = /** Runs `afterMeasurementMutation` on the rows it wrote. */ const TAIL = "runs afterMeasurementMutation"; +/** + * Runs the provider write step (`integrations/sync-kit/after-write.ts`), + * which is `afterMeasurementMutation` plus the arrivals and, where asked, the + * dashboard snapshot. + */ +const PROVIDER_TAIL = "runs afterProviderWrite"; + /** * Every file with a measurement write, and how it keeps the two readers * current. A file listed here must still contain a write (a stale entry @@ -57,17 +64,13 @@ const WRITE_SITES: Record = { // Retiring the equivalent-distance rows earlier Oura syncs wrote. "lib/oura/retire-distance.ts": TAIL, // Integration syncs: each collects the (type, day) pairs it touched and - // runs both legs itself at the end of the sync. - "lib/fitbit/sync-core.ts": - "refolds the touched days and re-warms the touched types at the end of the upsert", - "lib/google-health/sync-core.ts": - "refolds the touched days and re-warms the touched types at the end of the upsert", - "lib/nightscout/sync.ts": - "refolds the touched days and re-warms the touched types at the end of the sync", + // runs the provider write step over them. + "lib/fitbit/sync-core.ts": PROVIDER_TAIL, + "lib/google-health/sync-core.ts": PROVIDER_TAIL, + "lib/nightscout/sync.ts": PROVIDER_TAIL, + "lib/withings/sync.ts": PROVIDER_TAIL, "lib/oura/sync-heart-rate.ts": "collects the days each heart-rate page touched; refolds them and re-warms the type once per window", - "lib/withings/sync.ts": - "refolds the touched days and re-warms the touched types at the end of the sync", "lib/withings/sync-activity.ts": "refolds the touched days (and the day a moved row left) and re-warms the touched types", "lib/withings/sync-ecg.ts": @@ -168,6 +171,18 @@ describe("measurement write tail", () => { expect(missing).toEqual([]); }); + it("the files that claim the provider write step call it", () => { + const claimed = Object.entries(WRITE_SITES) + .filter(([, reason]) => reason === PROVIDER_TAIL) + .map(([file]) => file); + expect(claimed.length).toBeGreaterThan(0); + const missing = claimed.filter( + (file) => + !/afterProviderWrite\s*\(/.test(readFileSync(join(SRC, file), "utf8")), + ); + expect(missing).toEqual([]); + }); + it("the syncs that run both legs by hand call both", () => { const handRolled = Object.entries(WRITE_SITES) .filter(([, reason]) => reason.includes("re-warms")) diff --git a/src/__tests__/provider-ai-boundary-guard.test.ts b/src/__tests__/provider-ai-boundary-guard.test.ts new file mode 100644 index 000000000..01b8c2629 --- /dev/null +++ b/src/__tests__/provider-ai-boundary-guard.test.ts @@ -0,0 +1,70 @@ +/** + * A provider sync loads none of the AI layer. + * + * Thirteen sync modules once imported `insights/comprehensive-generate` to + * reach one invalidation function it re-exported, so every sync worker + * evaluated the whole insight generator. The write tail is + * `integrations/sync-kit/after-write.ts` now, and it reaches the status cards + * through `insights/status-invalidation`. This guard holds that: + * + * - no production module in a provider folder (`PROVIDER_DIRS`) imports + * anything under `lib/ai/` or the comprehensive generator; + * - and loading any of them does not evaluate the comprehensive generator + * transitively either. + * + * The planted case runs the transitive check over a module that does load + * it, so a walk that resolves nothing cannot pass. + */ +import { describe, expect, it } from "vitest"; + +import { PROVIDER_DIRS } from "./helpers/provider-dirs"; +import { requireFloor, walkSourceFiles } from "./helpers/source-files"; +import { + SRC, + moduleImports, + parse, + readSrc, + staticImportClosure, +} from "./helpers/source-metrics"; + +const GENERATOR = "lib/insights/comprehensive-generate.ts"; +const FORBIDDEN_SPEC = /^@\/lib\/(?:ai\/|insights\/comprehensive-generate$)/; + +function providerModules(): string[] { + return PROVIDER_DIRS.flatMap((dir) => + walkSourceFiles(`${SRC}/lib/${dir}`, { floor: 1 }) + .filter((f) => !f.includes("__tests__") && !/\.test\.tsx?$/.test(f)) + .map((f) => `lib/${dir}/${f}`), + ); +} + +describe("provider folders and the AI layer", () => { + const modules = providerModules(); + + it("reads every provider folder", () => { + requireFloor("provider modules", modules, 60); + }); + + it("imports nothing from lib/ai or the comprehensive generator", () => { + const offenders = modules.flatMap((rel) => + moduleImports(parse(rel, readSrc(rel))) + .value.filter((spec) => FORBIDDEN_SPEC.test(spec)) + .map((spec) => `${rel} → ${spec}`), + ); + expect(offenders).toEqual([]); + }); + + it("does not load the comprehensive generator transitively", () => { + const closure = staticImportClosure(modules); + expect(closure.size).toBeGreaterThan(modules.length); + expect(closure.has(GENERATOR)).toBe(false); + }); + + it("catches a module that loads it (planted)", () => { + expect( + staticImportClosure(["lib/jobs/morning-digest-refresh.ts"]).has( + GENERATOR, + ), + ).toBe(true); + }); +}); diff --git a/src/lib/fitbit/__tests__/poll-cohort.test.ts b/src/lib/fitbit/__tests__/poll-cohort.test.ts index c042846ec..16d8ef160 100644 --- a/src/lib/fitbit/__tests__/poll-cohort.test.ts +++ b/src/lib/fitbit/__tests__/poll-cohort.test.ts @@ -33,7 +33,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => ({ rowsUpserted: 0, durationMs: 0 })), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); diff --git a/src/lib/fitbit/__tests__/sync-dead-token.test.ts b/src/lib/fitbit/__tests__/sync-dead-token.test.ts index 23e68bdfe..d80c5d007 100644 --- a/src/lib/fitbit/__tests__/sync-dead-token.test.ts +++ b/src/lib/fitbit/__tests__/sync-dead-token.test.ts @@ -60,7 +60,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/logging/context", () => ({ diff --git a/src/lib/fitbit/__tests__/sync-plausibility.test.ts b/src/lib/fitbit/__tests__/sync-plausibility.test.ts index 1ba60d6fa..0d2fa95b0 100644 --- a/src/lib/fitbit/__tests__/sync-plausibility.test.ts +++ b/src/lib/fitbit/__tests__/sync-plausibility.test.ts @@ -35,7 +35,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/integrations/oauth-refresh", () => ({ diff --git a/src/lib/fitbit/__tests__/sync-sleep-replace.test.ts b/src/lib/fitbit/__tests__/sync-sleep-replace.test.ts index 4586a24c2..14a961fbe 100644 --- a/src/lib/fitbit/__tests__/sync-sleep-replace.test.ts +++ b/src/lib/fitbit/__tests__/sync-sleep-replace.test.ts @@ -60,7 +60,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/integrations/oauth-refresh", () => ({ diff --git a/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts b/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts index 7a3298379..9efdde5cc 100644 --- a/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts +++ b/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts @@ -48,7 +48,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/integrations/oauth-refresh", () => ({ diff --git a/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts index 2be8cada5..c1a2299f5 100644 --- a/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts +++ b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts @@ -73,7 +73,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: (...a: unknown[]) => recomputeUserRollups(...a), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); diff --git a/src/lib/fitbit/__tests__/sync.test.ts b/src/lib/fitbit/__tests__/sync.test.ts index effbc1506..d0e38d630 100644 --- a/src/lib/fitbit/__tests__/sync.test.ts +++ b/src/lib/fitbit/__tests__/sync.test.ts @@ -43,10 +43,11 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) => rows, recomputeBucketsForMeasurement: vi.fn(async () => {}), + recomputeBucketsForKeys: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => ({ rowsUpserted: 0, durationMs: 0 })), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index 2a2d6d57f..210f55c96 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/src/lib/fitbit/sync-core.ts @@ -10,16 +10,13 @@ import type { MeasurementType } from "@/generated/prisma/client"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { encrypt, decrypt } from "@/lib/crypto"; import { prisma } from "@/lib/db"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { persistRotatedToken } from "@/lib/integrations/oauth-refresh"; import { recordSyncFailure } from "@/lib/integrations/status"; import type { FailureKind } from "@/lib/integrations/status"; import { getEvent } from "@/lib/logging/context"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; +import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups"; +import { afterProviderWrite } from "@/lib/integrations/sync-kit/after-write"; import { fitbitOAuth } from "./oauth"; import { isFitbitRetired } from "./retirement"; import { getUserFitbitCredentials } from "./credentials"; @@ -898,23 +895,9 @@ export async function upsertFitbitMeasurements( return { imported, touched, inserted: insertedRows }; } - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `fitbit: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `fitbit: rollup recompute failed for ${userId}: ${err}`, - ); - } + // The arrivals went out per chunk above; the shared tail folds, marks the + // caches and re-warms the status cards. + await afterProviderWrite({ userId, provider: "fitbit", touched }); return { imported, touched, inserted: insertedRows }; } diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index 3e45ac27d..2a092bf86 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -2,10 +2,9 @@ import pLimit from "p-limit"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { prisma } from "@/lib/db"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status"; import { annotate, getEvent } from "@/lib/logging/context"; -import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { afterProviderBackfill } from "@/lib/integrations/sync-kit/after-write"; import { syncUserActivity } from "./sync-activity"; import { hardFailStorage, @@ -127,22 +126,10 @@ export async function syncUserFitbit( // + 3 queue sends). Best-effort: a populator hiccup never fails the backfill. if (opts.fullSync && deferTracker.keys.size > 0) { try { - const days = Array.from(deferTracker.keys.values()); - const types = Array.from(new Set(days.map((k) => k.type))); - const sorted = days - .map((k) => k.measuredAt.getTime()) - .sort((a, b) => a - b); - const from = new Date(sorted[0]!); - // The keys are UTC day-starts; extend `to` past the last touched day so - // the aggregator's `< to` upper bound covers it. - const to = new Date(sorted[sorted.length - 1]! + 24 * 60 * 60 * 1000); - await recomputeUserRollups(userId, { types, from, to }); - invalidateStatusInsightsForTypes(userId, types, { - source: "sync", - }).catch((err) => { - getEvent()?.addWarning( - `fitbit: status-insight invalidate failed for ${userId}: ${err}`, - ); + await afterProviderBackfill({ + userId, + provider: "fitbit", + keys: Array.from(deferTracker.keys.values()), }); } catch (err) { getEvent()?.addWarning( diff --git a/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts b/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts index cea735f92..9bbeca946 100644 --- a/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts +++ b/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts @@ -30,7 +30,7 @@ vi.mock("@/lib/integrations/status", () => ({ recordSyncFailure: vi.fn(async () => {}), recordSyncSuccess: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/arrivals/measurement-emit", () => ({ diff --git a/src/lib/google-health/__tests__/sync-failsoft.test.ts b/src/lib/google-health/__tests__/sync-failsoft.test.ts index d9684e8e7..46cb8d9cf 100644 --- a/src/lib/google-health/__tests__/sync-failsoft.test.ts +++ b/src/lib/google-health/__tests__/sync-failsoft.test.ts @@ -76,7 +76,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/cache/invalidate", () => ({ diff --git a/src/lib/google-health/__tests__/sync-fairness.test.ts b/src/lib/google-health/__tests__/sync-fairness.test.ts index 8f3aaf35f..5351336a3 100644 --- a/src/lib/google-health/__tests__/sync-fairness.test.ts +++ b/src/lib/google-health/__tests__/sync-fairness.test.ts @@ -66,7 +66,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: vi.fn(() => []), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("../sync-workout", () => ({ syncUserWorkout: resources.workout })); diff --git a/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts b/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts index 92e68d395..37386c3cb 100644 --- a/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts +++ b/src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts @@ -44,7 +44,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: vi.fn(() => []), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("../sync-workout", () => ({ syncUserWorkout: vi.fn(async () => 0) })); diff --git a/src/lib/google-health/__tests__/sync-outcomes.test.ts b/src/lib/google-health/__tests__/sync-outcomes.test.ts index 1aae53140..1f80ae21e 100644 --- a/src/lib/google-health/__tests__/sync-outcomes.test.ts +++ b/src/lib/google-health/__tests__/sync-outcomes.test.ts @@ -142,7 +142,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: vi.fn(() => []), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("../sync-core", () => ({ diff --git a/src/lib/google-health/__tests__/sync-plausibility.test.ts b/src/lib/google-health/__tests__/sync-plausibility.test.ts index 7b97e05c8..e5136e07b 100644 --- a/src/lib/google-health/__tests__/sync-plausibility.test.ts +++ b/src/lib/google-health/__tests__/sync-plausibility.test.ts @@ -35,7 +35,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("../credentials", () => ({ diff --git a/src/lib/google-health/__tests__/sync-sleep-replace.test.ts b/src/lib/google-health/__tests__/sync-sleep-replace.test.ts index bb5380027..180111464 100644 --- a/src/lib/google-health/__tests__/sync-sleep-replace.test.ts +++ b/src/lib/google-health/__tests__/sync-sleep-replace.test.ts @@ -26,7 +26,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("../credentials", () => ({ diff --git a/src/lib/google-health/__tests__/sync-time-budget.test.ts b/src/lib/google-health/__tests__/sync-time-budget.test.ts index 8b2bd3b4b..0ed7cebb0 100644 --- a/src/lib/google-health/__tests__/sync-time-budget.test.ts +++ b/src/lib/google-health/__tests__/sync-time-budget.test.ts @@ -46,7 +46,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: vi.fn(() => []), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("../sync-workout", () => ({ syncUserWorkout: resources.workout })); diff --git a/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts b/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts index 56b640bde..fc09f1ac8 100644 --- a/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts +++ b/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts @@ -39,7 +39,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("../credentials", () => ({ diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index 282a3a112..e32e3b5f9 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -38,11 +38,8 @@ import { import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups"; +import { afterProviderWrite } from "@/lib/integrations/sync-kit/after-write"; import { googleHealthOutcomeTruncated, googleHealthSliceExhausted, @@ -1553,24 +1550,14 @@ export async function upsertGoogleHealthMeasurements( return { imported, touched, inserted: insertedRows }; } - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `google-health: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `google-health: rollup recompute failed for ${userId}: ${err}`, - ); - noteHardFailure("rollup"); - } + // The arrivals went out per chunk above; the shared tail folds, marks the + // caches and re-warms the status cards. A failed fold fails the verdict. + const { rollupFailed } = await afterProviderWrite({ + userId, + provider: "google-health", + touched, + }); + if (rollupFailed) noteHardFailure("rollup"); noteGoogleHealthWritten(imported); return { imported, touched, inserted: insertedRows }; diff --git a/src/lib/google-health/sync.ts b/src/lib/google-health/sync.ts index 0caff675a..b26f32416 100644 --- a/src/lib/google-health/sync.ts +++ b/src/lib/google-health/sync.ts @@ -2,11 +2,7 @@ import pLimit from "p-limit"; import { prisma } from "@/lib/db"; import { annotate, getEvent } from "@/lib/logging/context"; import { recordSyncSuccess } from "@/lib/integrations/status"; -import { - collapseToTypeDayKeys, - recomputeUserRollups, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { afterProviderBackfill } from "@/lib/integrations/sync-kit/after-write"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { resolveUserTimezone } from "@/lib/tz/resolver"; import { @@ -362,20 +358,10 @@ async function runGoogleHealthSyncCycle( if (opts.fullSync && cycle.deferredRollupKeys.length > 0) { try { - const days = collapseToTypeDayKeys(cycle.deferredRollupKeys); - const types = Array.from(new Set(days.map((key) => key.type))); - const sorted = days - .map((key) => key.measuredAt.getTime()) - .sort((a, b) => a - b); - const from = new Date(sorted[0]!); - const to = new Date(sorted[sorted.length - 1]! + 24 * 60 * 60 * 1000); - await recomputeUserRollups(userId, { types, from, to }); - invalidateStatusInsightsForTypes(userId, types, { - source: "sync", - }).catch((err) => { - getEvent()?.addWarning( - `google-health: status-insight invalidate failed for ${userId}: ${err}`, - ); + await afterProviderBackfill({ + userId, + provider: "google-health", + keys: cycle.deferredRollupKeys, }); } catch (err) { anyFailed = true; diff --git a/src/lib/integrations/sync-kit/after-write.ts b/src/lib/integrations/sync-kit/after-write.ts new file mode 100644 index 000000000..99c0b542c --- /dev/null +++ b/src/lib/integrations/sync-kit/after-write.ts @@ -0,0 +1,140 @@ +/** + * The one step after a provider sync or import writes measurements. + * + * Every provider used to end its write with its own hand-written tail: fold + * the touched rollup days, drop the status cards of the touched types, mark + * the read caches stale, emit the data arrivals. The copies drifted (one + * skipped the status cards, one marked every cached block stale, one never + * marked anything), and each imported the whole insight generator to reach + * the invalidation it needed. `afterProviderWrite` is that tail, in one fixed + * order: + * + * 1. the freshly inserted rows go out on the data-arrival spine + * (fire and forget); + * 2. the touched `(type, day)` buckets are refolded, the read caches are + * marked stale back to the earliest touched reading (older closed blocks + * survive), and the status cards of the touched types are re-warmed + * (`afterMeasurementMutation`, shared with the interactive routes); + * 3. where the provider asks for it, the dashboard snapshot is evicted. + * + * Every leg is best effort: a populator hiccup never fails the sync. The + * result says whether the fold failed, for the providers whose cycle verdict + * counts it. + * + * A full-history backfill defers the per-day fold and refolds the whole + * touched range once at the end (`afterProviderBackfill`). + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { + emitInsertedMeasurementArrivals, + type InsertedMeasurementArrivalRow, +} from "@/lib/arrivals/measurement-emit"; +import { invalidateUserDashboardSnapshot } from "@/lib/cache/invalidate"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation"; +import { getEvent } from "@/lib/logging/context"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import { + collapseToTypeDayKeys, + recomputeUserRollups, +} from "@/lib/rollups/measurement-rollups"; + +type ArrivalSource = Parameters[2]; + +export interface TouchedReading { + type: MeasurementType; + measuredAt: Date; +} + +export interface ProviderWrite { + userId: string; + /** The provider, as the warn lines name it (`withings`, `google-health`). */ + provider: string; + /** Every reading the write created, changed or displaced. */ + touched: readonly TouchedReading[]; + /** Rows freshly inserted, for the arrival spine; omitted when none go out. */ + arrivals?: { + rows: InsertedMeasurementArrivalRow[]; + source: ArrivalSource; + }; + /** + * Re-warm the touched status cards now. A provider that writes in chunks + * and invalidates once at the end passes false and calls + * `invalidateProviderStatus` itself. + */ + invalidateStatus?: boolean; + /** Evict the dashboard snapshot as well (a sync with no client mutation). */ + evictDashboard?: boolean; +} + +export async function afterProviderWrite( + write: ProviderWrite, +): Promise<{ rollupFailed: boolean }> { + const { userId, provider, touched } = write; + if (write.arrivals) { + void emitInsertedMeasurementArrivals( + userId, + write.arrivals.rows, + write.arrivals.source, + ).catch(() => {}); + } + const { rollupFailed } = await afterMeasurementMutation( + userId, + [...touched], + { + source: "sync", + label: provider, + invalidateStatus: write.invalidateStatus, + }, + ); + if (rollupFailed) { + getEvent()?.addWarning( + `${provider}: rollup recompute failed for ${userId}`, + ); + } + if (write.evictDashboard && touched.length > 0) { + invalidateUserDashboardSnapshot(userId); + } + return { rollupFailed }; +} + +/** Re-warm the status cards of `types` after a sync (fire and forget). */ +export function invalidateProviderStatus( + userId: string, + provider: string, + types: Iterable, +): void { + const list = [...new Set(types)]; + if (list.length === 0) return; + void invalidateStatusInsightsForTypes(userId, list, { + source: "sync", + }).catch((err) => { + getEvent()?.addWarning( + `${provider}: status-insight invalidate failed for ${userId}: ${err}`, + ); + }); +} + +/** + * The end of a full-history backfill: refold the whole touched range once + * (from the first touched day to the end of the last), then re-warm the + * touched status cards. Throws when the fold fails, for the caller to count. + */ +export async function afterProviderBackfill(input: { + userId: string; + provider: string; + /** The deferred `(type, day)` keys the backfill's writes collected. */ + keys: readonly TouchedReading[]; +}): Promise { + const days = collapseToTypeDayKeys([...input.keys]); + if (days.length === 0) return; + const types = Array.from(new Set(days.map((key) => key.type))); + const sorted = days + .map((key) => key.measuredAt.getTime()) + .sort((a, b) => a - b); + const from = new Date(sorted[0]!); + // The keys are UTC day starts; the end is past the last touched day so the + // aggregator's `< to` bound covers it. + const to = new Date(sorted[sorted.length - 1]! + 24 * 60 * 60 * 1000); + await recomputeUserRollups(input.userId, { types, from, to }); + invalidateProviderStatus(input.userId, input.provider, types); +} diff --git a/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts b/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts index bbb619a11..23922aee7 100644 --- a/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts +++ b/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts @@ -50,7 +50,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/integrations/oauth-refresh", () => ({ diff --git a/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts b/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts index cf6c36c81..f4d56ca51 100644 --- a/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts +++ b/src/lib/jobs/__tests__/google-health-sleep-repair.test.ts @@ -64,7 +64,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ recomputeBucketsForMeasurement: vi.fn(async () => {}), recomputeUserRollups: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/google-health/credentials", () => ({ diff --git a/src/lib/nightscout/__tests__/sync-plausibility.test.ts b/src/lib/nightscout/__tests__/sync-plausibility.test.ts index 24c023eb7..eb8b8942a 100644 --- a/src/lib/nightscout/__tests__/sync-plausibility.test.ts +++ b/src/lib/nightscout/__tests__/sync-plausibility.test.ts @@ -32,9 +32,10 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) => rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), recomputeBucketsForMeasurement: recomputeMock, + recomputeBucketsForKeys: recomputeMock, })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: invalidateMock, })); diff --git a/src/lib/nightscout/__tests__/sync.test.ts b/src/lib/nightscout/__tests__/sync.test.ts index 18f3c8a9b..d5aa762e3 100644 --- a/src/lib/nightscout/__tests__/sync.test.ts +++ b/src/lib/nightscout/__tests__/sync.test.ts @@ -51,7 +51,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ }, })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: invalidateMock, })); diff --git a/src/lib/nightscout/sync.ts b/src/lib/nightscout/sync.ts index fa3998a0b..f6d0728fd 100644 --- a/src/lib/nightscout/sync.ts +++ b/src/lib/nightscout/sync.ts @@ -32,16 +32,9 @@ import { } from "@/lib/integrations/status"; import type { IntegrationClassification } from "@/lib/integrations/http-status-classifier"; import type { SyncWriteResult } from "@/lib/outcome/written-outcome"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { - emitInsertedMeasurementArrivals, - type InsertedMeasurementArrivalRow, -} from "@/lib/arrivals/measurement-emit"; +import { afterProviderWrite } from "@/lib/integrations/sync-kit/after-write"; +import type { InsertedMeasurementArrivalRow } from "@/lib/arrivals/measurement-emit"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { fetchSgvEntries, mapSgvEntryToMeasurement, @@ -319,29 +312,12 @@ export async function upsertNightscoutEntries( // inert for glucose today — it removes Nightscout as the spine's odd one out // and lights up automatically once a `glucose` arrival kind + its consuming // surface land in a later release. - void emitInsertedMeasurementArrivals( + await afterProviderWrite({ userId, - insertedRows, - "nightscout", - ).catch(() => {}); - - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `nightscout: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `nightscout: rollup fold failed for ${userId}: ${err}`, - ); - } + provider: "nightscout", + touched, + arrivals: { rows: insertedRows, source: "nightscout" }, + }); return { inserted: insertedRows.length, failedRows }; } diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index 6c9401d76..cfaa1ac78 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/src/lib/oura/__tests__/sync.test.ts @@ -125,7 +125,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ }, })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: invalidateMock, })); diff --git a/src/lib/oura/__tests__/workouts-heart-rate.test.ts b/src/lib/oura/__tests__/workouts-heart-rate.test.ts index 7460e99f1..9d2bd56ea 100644 --- a/src/lib/oura/__tests__/workouts-heart-rate.test.ts +++ b/src/lib/oura/__tests__/workouts-heart-rate.test.ts @@ -33,7 +33,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({ >()), recomputeBucketsForKeys: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index fe4f89984..804b15e06 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -45,12 +45,10 @@ import { toFailureKind, type FailureKind, } from "@/lib/integrations/status"; -import { - emitInsertedMeasurementArrivals, - type InsertedMeasurementArrivalRow, -} from "@/lib/arrivals/measurement-emit"; +import type { InsertedMeasurementArrivalRow } from "@/lib/arrivals/measurement-emit"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import { afterProviderWrite } from "@/lib/integrations/sync-kit/after-write"; import type { SyncWriteResult } from "@/lib/outcome/written-outcome"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; import { @@ -885,17 +883,17 @@ export async function upsertOuraMeasurements( } opts.onInserted?.(insertedRows); - if (opts.emitArrivals !== false) { - void emitInsertedMeasurementArrivals(userId, insertedRows, "oura").catch( - () => {}, - ); - } - // 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", + // The shared tail: arrivals (unless the caller emits its own), 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 afterProviderWrite({ + userId, + provider: "oura", + touched, + ...(opts.emitArrivals !== false + ? { arrivals: { rows: insertedRows, source: "oura" as const } } + : {}), }); return imported; diff --git a/src/lib/polar/__tests__/sync.test.ts b/src/lib/polar/__tests__/sync.test.ts index b4d36ef63..9681c8a58 100644 --- a/src/lib/polar/__tests__/sync.test.ts +++ b/src/lib/polar/__tests__/sync.test.ts @@ -69,9 +69,10 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) => rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), recomputeBucketsForMeasurement: recomputeMock, + recomputeBucketsForKeys: recomputeMock, })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: invalidateMock, })); diff --git a/src/lib/polar/sync.ts b/src/lib/polar/sync.ts index 080108b79..f48aafcb1 100644 --- a/src/lib/polar/sync.ts +++ b/src/lib/polar/sync.ts @@ -53,18 +53,11 @@ import { toFailureKind, type FailureKind, } from "@/lib/integrations/status"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; +import { afterProviderWrite } from "@/lib/integrations/sync-kit/after-write"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import type { SyncWriteResult } from "@/lib/outcome/written-outcome"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; -import { - emitInsertedMeasurementArrivals, - type InsertedMeasurementArrivalRow, -} from "@/lib/arrivals/measurement-emit"; +import type { InsertedMeasurementArrivalRow } from "@/lib/arrivals/measurement-emit"; import { MeasurementReconciliationError, reconcileExternalMeasurement, @@ -441,26 +434,12 @@ export async function upsertPolarMeasurements( } opts.onInserted?.(insertedRows); - void emitInsertedMeasurementArrivals(userId, insertedRows, "polar").catch( - () => {}, - ); - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `polar: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - } catch (err) { - getEvent()?.addWarning( - `polar: rollup recompute failed for ${userId}: ${err}`, - ); - } + await afterProviderWrite({ + userId, + provider: "polar", + touched, + arrivals: { rows: insertedRows, source: "polar" }, + }); return imported; } diff --git a/src/lib/rollups/after-measurement-mutation.ts b/src/lib/rollups/after-measurement-mutation.ts index c96bbecdf..6cd1f834e 100644 --- a/src/lib/rollups/after-measurement-mutation.ts +++ b/src/lib/rollups/after-measurement-mutation.ts @@ -53,17 +53,24 @@ export async function afterMeasurementMutation( source: StatusIngestSource; /** Log prefix identifying the calling surface in the warn lines. */ label?: string; + /** + * Re-warm the status cards of the touched types. A caller that writes in + * chunks and invalidates once at the end passes false. + */ + invalidateStatus?: boolean; }, -): Promise { +): Promise<{ rollupFailed: boolean }> { const label = options.label ?? "measurements"; - if (rows.length === 0) return; + if (rows.length === 0) return { rollupFailed: false }; // Collapse to the distinct `(type, day)` set BEFORE recomputing so a // 500-row batch fires at most one DAY recompute per touched bucket. const keys = collapseToTypeDayKeys(rows); + let rollupFailed = false; try { await recomputeBucketsForKeys(userId, keys); } catch (err) { + rollupFailed = true; console.warn(`[${label}] rollup recompute failed`, err); } invalidateUserMeasurements(userId, { touchedFrom: earliestMeasuredAt(rows) }); @@ -71,7 +78,7 @@ export async function afterMeasurementMutation( // 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. const types = Array.from(new Set(keys.map((k) => k.type))); - if (types.length > 0) { + if (types.length > 0 && options.invalidateStatus !== false) { // Registered so the test suite can await it: the warn below is a console // write on a detached handle, and one that lands while a vitest worker is // closing its RPC channel fails the whole run. No-op outside test. @@ -83,4 +90,5 @@ export async function afterMeasurementMutation( }), ); } + return { rollupFailed }; } diff --git a/src/lib/whoop/__tests__/sync-dead-token.test.ts b/src/lib/whoop/__tests__/sync-dead-token.test.ts index a5fadba59..46f676086 100644 --- a/src/lib/whoop/__tests__/sync-dead-token.test.ts +++ b/src/lib/whoop/__tests__/sync-dead-token.test.ts @@ -71,7 +71,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: vi.fn(() => []), recomputeBucketsForMeasurement: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); vi.mock("@/lib/arrivals/measurement-emit", () => ({ diff --git a/src/lib/whoop/__tests__/sync-user-whoop.test.ts b/src/lib/whoop/__tests__/sync-user-whoop.test.ts index 2e4c30e30..c74c91b04 100644 --- a/src/lib/whoop/__tests__/sync-user-whoop.test.ts +++ b/src/lib/whoop/__tests__/sync-user-whoop.test.ts @@ -55,7 +55,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: () => [], recomputeBucketsForMeasurement: vi.fn(async () => {}), })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn(async () => {}), })); diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index f989d22a0..e05221176 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -108,7 +108,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({ }, })); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: (...a: unknown[]) => invalidateStatusMock(...a), })); diff --git a/src/lib/whoop/sync-core.ts b/src/lib/whoop/sync-core.ts index 48f88adcb..86bf44443 100644 --- a/src/lib/whoop/sync-core.ts +++ b/src/lib/whoop/sync-core.ts @@ -34,14 +34,10 @@ import { type FailureKind, } from "@/lib/integrations/status"; import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; -import { - emitInsertedMeasurementArrivals, - type InsertedMeasurementArrivalRow, -} from "@/lib/arrivals/measurement-emit"; + afterProviderWrite, + invalidateProviderStatus, +} from "@/lib/integrations/sync-kit/after-write"; +import type { InsertedMeasurementArrivalRow } from "@/lib/arrivals/measurement-emit"; import { obtainAccessToken } from "@/lib/integrations/oauth-kit/token"; import { MeasurementReconciliationError, @@ -351,17 +347,8 @@ export async function upsertWhoopMeasurements( let imported = 0; const touchedTypes = new Set(); - const invalidateTouchedTypes = (): void => { - if (touchedTypes.size === 0) return; - - void invalidateStatusInsightsForTypes(userId, [...touchedTypes], { - source: "sync", - }).catch((err) => { - getEvent()?.addWarning( - `whoop: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - }; + const invalidateTouchedTypes = (): void => + invalidateProviderStatus(userId, "whoop", touchedTypes); for ( let chunkStart = 0; @@ -431,17 +418,15 @@ export async function upsertWhoopMeasurements( } opts.onInserted?.(insertedRows); - void emitInsertedMeasurementArrivals(userId, insertedRows, "whoop").catch( - () => {}, - ); - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - } catch (err) { - getEvent()?.addWarning( - `whoop: rollup recompute failed for ${userId}: ${err}`, - ); - } + // Each chunk folds its own days; the status cards are re-warmed once, + // after the last chunk (or the one that failed). + await afterProviderWrite({ + userId, + provider: "whoop", + touched, + arrivals: { rows: insertedRows, source: "whoop" }, + invalidateStatus: false, + }); } invalidateTouchedTypes(); diff --git a/src/lib/withings/__tests__/subscription-state.test.ts b/src/lib/withings/__tests__/subscription-state.test.ts index cd6718b60..079a2ca6f 100644 --- a/src/lib/withings/__tests__/subscription-state.test.ts +++ b/src/lib/withings/__tests__/subscription-state.test.ts @@ -70,7 +70,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { }; }); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), })); diff --git a/src/lib/withings/__tests__/sync-ecg.test.ts b/src/lib/withings/__tests__/sync-ecg.test.ts index 916677346..807efb3e3 100644 --- a/src/lib/withings/__tests__/sync-ecg.test.ts +++ b/src/lib/withings/__tests__/sync-ecg.test.ts @@ -84,10 +84,6 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { }; }); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ - invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), -})); - import { prisma } from "@/lib/db"; import { isReauthRequired, diff --git a/src/lib/withings/__tests__/sync-measure-watermark.test.ts b/src/lib/withings/__tests__/sync-measure-watermark.test.ts index 1a3397c9b..ff2317418 100644 --- a/src/lib/withings/__tests__/sync-measure-watermark.test.ts +++ b/src/lib/withings/__tests__/sync-measure-watermark.test.ts @@ -74,7 +74,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { }; }); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), })); @@ -253,11 +253,11 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () expect(recordSyncFailure).not.toHaveBeenCalled(); }); - // Watched red: with the `invalidateUserMeasurements(userId)` call removed - // from the sync tail this fails — the pre-fix tail evicted only the - // dashboard snapshot, so correlations / analytics / targets / derived - // served pre-sync bodies until their TTL lapsed after every hourly sync - // (the correlations route documents the invariant this restores). + // Without the cache leg of the shared write tail this fails: an earlier + // tail evicted only the dashboard snapshot, so correlations, analytics, + // targets and derived served pre-sync bodies until their TTL lapsed after + // every hourly sync. The caches are marked stale back to the earliest + // reading the sync wrote, so closed blocks of older readings survive. it("marks the per-user measurement caches stale when rows landed", async () => { vi.mocked(fetchMeasurements).mockResolvedValue([ { @@ -273,6 +273,7 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () await import("@/lib/cache/invalidate"); expect(vi.mocked(invalidateUserMeasurements)).toHaveBeenCalledWith( "user-1", + { touchedFrom: new Date("2026-05-16T08:00:00Z") }, ); }); }); diff --git a/src/lib/withings/__tests__/sync-plan.test.ts b/src/lib/withings/__tests__/sync-plan.test.ts index 1cf80a2a5..e63259e97 100644 --- a/src/lib/withings/__tests__/sync-plan.test.ts +++ b/src/lib/withings/__tests__/sync-plan.test.ts @@ -65,7 +65,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { }; }); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), })); diff --git a/src/lib/withings/__tests__/sync-plausibility.test.ts b/src/lib/withings/__tests__/sync-plausibility.test.ts index 3f4a5e4c9..60901ebb6 100644 --- a/src/lib/withings/__tests__/sync-plausibility.test.ts +++ b/src/lib/withings/__tests__/sync-plausibility.test.ts @@ -76,7 +76,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { }; }); -vi.mock("@/lib/insights/comprehensive-generate", () => ({ +vi.mock("@/lib/insights/status-invalidation", () => ({ invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), })); diff --git a/src/lib/withings/sync.ts b/src/lib/withings/sync.ts index 86d780388..c49f0838e 100644 --- a/src/lib/withings/sync.ts +++ b/src/lib/withings/sync.ts @@ -26,7 +26,6 @@ import { type WithinPlanResult, } from "./api-plan"; import { annotate, getEvent } from "@/lib/logging/context"; -import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; import { isReauthRequired, @@ -39,15 +38,7 @@ import { WithingsApiError, type WithingsClassification, } from "./response-classifier"; -import { - collapseToTypeDayKeys, - recomputeBucketsForKeys, -} from "@/lib/rollups/measurement-rollups"; -import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; -import { - invalidateUserDashboardSnapshot, - invalidateUserMeasurements, -} from "@/lib/cache/invalidate"; +import { afterProviderWrite } from "@/lib/integrations/sync-kit/after-write"; import type { SyncWriteResult } from "@/lib/outcome/written-outcome"; import { obtainAccessToken } from "@/lib/integrations/oauth-kit/token"; import { withingsTokenDescriptor } from "./oauth-descriptor"; @@ -430,53 +421,17 @@ export async function syncUserMeasurements( } } - void emitInsertedMeasurementArrivals( + // The shared tail: arrivals, the rollup fold, the read caches back to the + // earliest touched reading and the touched status cards. The dashboard + // snapshot is evicted too: this server-side sync produces no client + // mutation event, so its cached body would otherwise outlive the import. + await afterProviderWrite({ userId, - insertedArrivals, - "withings", - ).catch(() => {}); - - // v1.4.39.1 — refresh the persistent rollup table for every distinct - // (type, day) the sync touched. The chart-data + analytics read paths - // consume these buckets via `source=rollup`; without this hook, - // Withings-only metrics stayed off the rollup tier and the dashboard - // chart at the 30-day range under-counted recent days. - try { - const keys = collapseToTypeDayKeys(touched); - await recomputeBucketsForKeys(userId, keys); - - // v1.8.0 — a Withings sync that lands new weight / BP / pulse / - // body-comp rows dirties the matching per-metric assessment caches. - // Drop them so the next mount / nightly warm pass regenerates - // against the new data. Fire-and-forget: never fails the sync. - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - { source: "sync" }, - ).catch((err) => { - getEvent()?.addWarning( - `withings: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); - - // v1.18.9 (#38) — hard-evict the dashboard snapshot so a focus-refetch - // after this server-side sync returns the freshly-imported readings. - // The server-side Withings path produces no client mutation event, so - // without this the snapshot's `cachedSwr` entry kept serving the - // pre-sync body until its ~180 s TTL lapsed. - if (keys.length > 0) { - invalidateUserDashboardSnapshot(userId); - // Background-sync posture (mirrors the Fitbit tail): mark the - // per-user analytics / correlations / targets / achievements cells - // stale so the imported rows reach the cached readers before their - // TTL lapses — the correlations route documents this invariant. - invalidateUserMeasurements(userId); - } - } catch (err) { - getEvent()?.addWarning( - `withings: rollup recompute failed for ${userId}: ${err}`, - ); - } + provider: "withings", + touched, + arrivals: { rows: insertedArrivals, source: "withings" }, + evictDashboard: true, + }); // v1.28.39 — HOLD the watermark on any hard row failure. Implements the // documented status-tracking contract above ("downstream measurement-upsert diff --git a/tests/integration/provider-after-write.test.ts b/tests/integration/provider-after-write.test.ts new file mode 100644 index 000000000..8ee2d3c54 --- /dev/null +++ b/tests/integration/provider-after-write.test.ts @@ -0,0 +1,99 @@ +/** + * The provider write step against the real schema: what a sync leaves behind + * after `afterProviderWrite` and `afterProviderBackfill`. + * + * - the touched days are folded into the DAY rollup the read paths use; + * - a write that touched nothing folds nothing and reports no failure; + * - a backfill refolds the whole touched range once, including a day that + * was written without the per-write fold. + * + * The data-arrival send and the status re-warm are queue work (the boss is + * absent here, so the arrival send is a no-op); they are pinned by the unit + * tests of the providers. What matters here is what Postgres holds. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { + afterProviderBackfill, + afterProviderWrite, +} from "@/lib/integrations/sync-kit/after-write"; + +import { makeMeasurement, makeUser } from "../kit/factories"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +beforeEach(() => truncateAllTables(getPrismaClient())); + +async function dayRollups(userId: string) { + return getPrismaClient().measurementRollup.findMany({ + where: { userId, granularity: "DAY", type: "WEIGHT" }, + select: { bucketStart: true, count: true, mean: true, source: true }, + orderBy: { bucketStart: "asc" }, + }); +} + +describe("provider write step", () => { + it("folds the days a sync touched", async () => { + const user = await makeUser(); + const first = await makeMeasurement(user, { + source: "WITHINGS", + value: 80, + measuredAt: new Date("2026-03-02T07:30:00Z"), + }); + const second = await makeMeasurement(user, { + source: "WITHINGS", + value: 82, + measuredAt: new Date("2026-03-02T19:30:00Z"), + }); + + const result = await afterProviderWrite({ + userId: user.id, + provider: "withings", + touched: [first, second], + evictDashboard: true, + }); + + expect(result).toEqual({ rollupFailed: false }); + const rows = await dayRollups(user.id); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ count: 2, mean: 81, source: "WITHINGS" }); + }); + + it("does nothing for a write that touched nothing", async () => { + const user = await makeUser(); + await expect( + afterProviderWrite({ userId: user.id, provider: "polar", touched: [] }), + ).resolves.toEqual({ rollupFailed: false }); + expect(await dayRollups(user.id)).toEqual([]); + }); + + it("refolds a backfill's whole range once at the end", async () => { + const user = await makeUser(); + const days = ["2026-03-01", "2026-03-04", "2026-03-09"]; + const written = []; + for (const day of days) { + written.push( + await makeMeasurement(user, { + source: "GOOGLE_HEALTH", + value: 70, + measuredAt: new Date(`${day}T08:00:00Z`), + }), + ); + } + + await afterProviderBackfill({ + userId: user.id, + provider: "google-health", + keys: written.map((row) => ({ + type: row.type, + measuredAt: new Date( + `${row.measuredAt.toISOString().slice(0, 10)}T00:00:00Z`, + ), + })), + }); + + const rows = await dayRollups(user.id); + expect( + rows.map((row) => row.bucketStart.toISOString().slice(0, 10)), + ).toEqual(days); + }); +});