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.
This commit is contained in:
Marc-André Bombeck committed 2026-10-23 13:12:23 +02:00
1 parent b6f45c0d88
commit e626f2123f
48 files changed
+474 -307

No files matched your search

@@ -1,7 +1,7 @@
{
"$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
"count": 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
}
}
@@ -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
}
}
@@ -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,
@@ -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<string, string> = {
// 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"))
@@ -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);
});
});
+1 -1
View File
@@ -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 () => {}),
}));
@@ -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", () => ({
@@ -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", () => ({
@@ -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", () => ({
@@ -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", () => ({
@@ -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 () => {}),
}));
+2 -1
View File
@@ -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 () => {}),
}));
+5 -22
View File
@@ -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 };
}
+5 -18
View File
@@ -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(
@@ -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", () => ({
@@ -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", () => ({
@@ -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 }));
@@ -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) }));
@@ -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", () => ({
@@ -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", () => ({
@@ -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", () => ({
@@ -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 }));
@@ -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", () => ({
+10 -23
View File
@@ -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 };
+5 -19
View File
@@ -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;
@@ -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<typeof emitInsertedMeasurementArrivals>[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<MeasurementType>,
): 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<void> {
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);
}
@@ -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", () => ({
@@ -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", () => ({
@@ -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,
}));
+1 -1
View File
@@ -51,7 +51,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
},
}));
vi.mock("@/lib/insights/comprehensive-generate", () => ({
vi.mock("@/lib/insights/status-invalidation", () => ({
invalidateStatusInsightsForTypes: invalidateMock,
}));
+7 -31
View File
@@ -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 };
}
+1 -1
View File
@@ -125,7 +125,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
},
}));
vi.mock("@/lib/insights/comprehensive-generate", () => ({
vi.mock("@/lib/insights/status-invalidation", () => ({
invalidateStatusInsightsForTypes: invalidateMock,
}));
@@ -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 () => {}),
}));
+13 -15
View File
@@ -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;
+2 -1
View File
@@ -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,
}));
+8 -29
View File
@@ -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;
}
+11 -3
View File
@@ -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<void> {
): 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 };
}
@@ -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", () => ({
@@ -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 () => {}),
}));
+1 -1
View File
@@ -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),
}));
+15 -30
View File
@@ -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<MeasurementType>();
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();
@@ -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),
}));
@@ -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,
@@ -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") },
);
});
});
+1 -1
View File
@@ -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),
}));
@@ -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),
}));
+11 -56
View File
@@ -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
@@ -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);
});
});