fix(withings-sync): fold persistent rollup tier on every touched (type, day)

Three Withings ingest paths wrote measurements without ever firing the
rollup write hook: `syncUserMeasurements`, `syncUserActivity`, and
`syncUserSleep`. Once the dashboard chart's `source=rollup` fast-path
landed in v1.4.36 the under-counting started biting Withings-only
metrics on the 30 / 90-day range — Marc's BP / pulse charts painted
the empty-state at 30 even though the live `measurements` table held
the days. The 7-day window kept reading raw rows so the empty-state
only appeared above the chart's `< 3 daily points` threshold.

Each sync now collects every `(type, measuredAt)` it touched, collapses
to distinct `(type, dayStart)` via `collapseToTypeDayKeys`, and calls
`recomputeBucketsForMeasurement` once per pair after the row writes.
The per-day collapse keeps a 30-day catch-up bounded — at most ~30
DAY recomputes per sync rather than one per row. Best-effort: a
populator hiccup never fails the user's sync.

- Tests pin the hook call count on activity (6 calls for 2 days × 3
  metrics, 3 calls when the dates collapse) and sleep (one per night
  even when each night has multiple stage segments), plus a populator-
  failure path that proves the sync still returns its imported count.
- Reproducer verified: stashing the sync-activity hook makes the new
  test red on develop; restoring it greens.
This commit is contained in:
Marc-André Bombeck committed 2026-05-21 09:01:35 +02:00
1 parent 4d895eeab0
commit ad22fa2e9d
5 files changed
+217

No files matched your search

@@ -45,12 +45,27 @@ vi.mock("@/lib/logging/context", () => ({
})),
}));
// v1.4.39.1 — the sync now hands its touched (type, day) keys to the
// measurement rollup writer so the dashboard chart's `source=rollup`
// fast-path sees the new buckets. The mock collects the call args so
// the regression test below can assert the hook fired.
vi.mock("@/lib/measurements/rollups", async () => {
const actual = await vi.importActual<
typeof import("@/lib/measurements/rollups")
>("@/lib/measurements/rollups");
return {
...actual,
recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
};
});
import { prisma } from "@/lib/db";
import {
parkIntegrationAtReauth,
recordSyncFailure,
recordSyncSuccess,
} from "@/lib/integrations/status";
import { recomputeBucketsForMeasurement } from "@/lib/measurements/rollups";
import { fetchWithingsActivity, syncUserActivity } from "../sync-activity";
@@ -363,3 +378,66 @@ describe("syncUserActivity — scope-skip guard (v1.4.26)", () => {
expect(parkIntegrationAtReauth).not.toHaveBeenCalled();
});
});
describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => {
it("folds the persistent rollup table for each (type, day) the sync touched", async () => {
// Two distinct calendar days × three activity metrics each = 6
// writes; the rollup hook should be called once per distinct
// `(type, dayStart)` so the dashboard chart's `source=rollup` fast-
// path lights up the Withings-ingested days. Pre-v1.4.39.1 the
// sync skipped the hook entirely, leaving the rollup tier blind to
// Withings activity.
installFetchMock([
{ date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
{ date: "2026-05-13", steps: 9100, distance: 7100, calories: 430 },
]);
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
await syncUserActivity("user-1");
expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(6);
const calls = vi
.mocked(recomputeBucketsForMeasurement)
.mock.calls.map((c) => ({
userId: c[0],
type: c[1],
measuredAtIso: (c[2] as Date).toISOString(),
}));
// Distinct (type, day) pairs the sync touched, derived from the two
// dates × three metrics × `collapseToTypeDayKeys`.
const distinct = new Set(calls.map((c) => `${c.type}|${c.measuredAtIso}`));
expect(distinct.size).toBe(6);
expect(calls.every((c) => c.userId === "user-1")).toBe(true);
});
it("collapses multiple writes for the same (type, day) into a single rollup recompute", async () => {
// A single-date entry exercises the existing day-collapse helper —
// three metric rows for 2026-05-12 must fold into three rollup
// calls (one per type), not nine.
installFetchMock([
{ date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
]);
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
await syncUserActivity("user-1");
expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(3);
});
it("swallows a populator failure so the sync still returns its imported count", async () => {
// The rollup recompute is a best-effort cache tier — a populator
// hiccup must never fail the user's sync. The sync's `imported`
// return value should still reflect the rows written.
installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
vi.mocked(recomputeBucketsForMeasurement).mockRejectedValueOnce(
new Error("simulated rollup failure"),
);
const imported = await syncUserActivity("user-1");
expect(imported).toBe(1);
});
});
@@ -44,12 +44,27 @@ vi.mock("@/lib/logging/context", () => ({
})),
}));
// v1.4.39.1 — the sleep sync now folds the persistent measurement
// rollup tier for each touched SLEEP_DURATION day so the dashboard
// chart's `source=rollup` fast-path sees the new segments. The mock
// surfaces the call args for the regression test below.
vi.mock("@/lib/measurements/rollups", async () => {
const actual = await vi.importActual<
typeof import("@/lib/measurements/rollups")
>("@/lib/measurements/rollups");
return {
...actual,
recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
};
});
import { prisma } from "@/lib/db";
import {
parkIntegrationAtReauth,
recordSyncFailure,
recordSyncSuccess,
} from "@/lib/integrations/status";
import { recomputeBucketsForMeasurement } from "@/lib/measurements/rollups";
import {
fetchWithingsSleep,
@@ -352,3 +367,47 @@ describe("syncUserSleep — scope-skip guard (v1.4.26)", () => {
expect(parkIntegrationAtReauth).not.toHaveBeenCalled();
});
});
describe("syncUserSleep — measurement rollup hook (v1.4.39.1)", () => {
it("folds the persistent rollup table once per night the sync touched", async () => {
// Two distinct nights, each with multiple stage segments. The
// sync should collapse the per-night segments to one rollup
// recompute per night via `collapseToTypeDayKeys`. Pre-v1.4.39.1
// the rollup tier never heard about Withings sleep at all.
const nightA = Math.floor(Date.UTC(2026, 4, 12, 22) / 1000); // 2026-05-12 22:00 UTC
const nightB = Math.floor(Date.UTC(2026, 4, 13, 22) / 1000); // 2026-05-13 22:00 UTC
installFetchMock([
{ startdate: nightA, enddate: nightA + 3600, state: 1, id: 1 },
{ startdate: nightA + 3600, enddate: nightA + 5400, state: 2, id: 1 },
{ startdate: nightA + 5400, enddate: nightA + 7200, state: 3, id: 1 },
{ startdate: nightB, enddate: nightB + 3600, state: 1, id: 2 },
{ startdate: nightB + 3600, enddate: nightB + 5400, state: 2, id: 2 },
]);
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
await syncUserSleep("user-1");
// Both segments of night A land in the same UTC day; night B in
// the next. Distinct (SLEEP_DURATION, dayStart) tuples = 2.
expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(2);
const types = vi
.mocked(recomputeBucketsForMeasurement)
.mock.calls.map((c) => c[1]);
expect(new Set(types)).toEqual(new Set(["SLEEP_DURATION"]));
});
it("swallows a populator failure so the sync still returns its imported count", async () => {
installFetchMock([
{ startdate: 1715000000, enddate: 1715003600, state: 2, id: 1 },
]);
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
vi.mocked(recomputeBucketsForMeasurement).mockRejectedValueOnce(
new Error("simulated rollup failure"),
);
const imported = await syncUserSleep("user-1");
expect(imported).toBe(1);
});
});
+24
View File
@@ -38,6 +38,10 @@ import type { MeasurementType } from "@/generated/prisma/client";
import { prisma } from "@/lib/db";
import { getEvent } from "@/lib/logging/context";
import { getUnitForType } from "@/lib/validations/measurement";
import {
collapseToTypeDayKeys,
recomputeBucketsForMeasurement,
} from "@/lib/measurements/rollups";
import { hasActivityScope } from "./client";
import {
@@ -285,6 +289,11 @@ export async function syncUserActivity(
}
let imported = 0;
// v1.4.39.1 — track every (type, measuredAt) we touched so the
// persistent rollup tier can be re-folded at the end of the sync.
// See sync.ts header for the full rationale; the chart's
// `source=rollup` fast-path otherwise misses Withings activity days.
const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
for (const entry of entries) {
if (!entry.date) continue;
const measuredAt = activityMeasuredAt(entry.date);
@@ -323,6 +332,7 @@ export async function syncUserActivity(
},
});
}
touched.push({ type, measuredAt });
imported++;
} catch (err) {
getEvent()?.addWarning(
@@ -332,6 +342,20 @@ export async function syncUserActivity(
}
}
// v1.4.39.1 — refresh the persistent rollup table for every distinct
// (type, day) the sync touched. Collapsed so a multi-day Withings
// catch-up costs at most ~N (type, day) recomputes.
try {
const keys = collapseToTypeDayKeys(touched);
for (const k of keys) {
await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
}
} catch (err) {
getEvent()?.addWarning(
`withings activity: rollup recompute failed for ${userId}: ${err}`,
);
}
await recordSyncSuccess(userId, "withings");
return imported;
}
+23
View File
@@ -36,6 +36,10 @@ import type { MeasurementType, SleepStage } from "@/generated/prisma/client";
import { prisma } from "@/lib/db";
import { getEvent } from "@/lib/logging/context";
import { getUnitForType } from "@/lib/validations/measurement";
import {
collapseToTypeDayKeys,
recomputeBucketsForMeasurement,
} from "@/lib/measurements/rollups";
import { hasActivityScope } from "./client";
import {
@@ -223,6 +227,10 @@ export async function syncUserSleep(
let imported = 0;
let segmentIndex = 0;
// v1.4.39.1 — track every (type, measuredAt) we touched so the
// persistent rollup tier can be re-folded at the end. See sync.ts for
// the full rationale.
const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
for (const segment of segments) {
const stage = mapWithingsSleepState(segment.state);
if (!stage) {
@@ -270,6 +278,7 @@ export async function syncUserSleep(
},
});
}
touched.push({ type: SLEEP_TYPE, measuredAt });
imported++;
} catch (err) {
getEvent()?.addWarning(
@@ -279,6 +288,20 @@ export async function syncUserSleep(
segmentIndex++;
}
// v1.4.39.1 — refresh the persistent rollup table for every distinct
// (type, day) the sync touched. Sleep segments collapse heavily —
// ~10 stage rows per night land in one DAY recompute.
try {
const keys = collapseToTypeDayKeys(touched);
for (const k of keys) {
await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
}
} catch (err) {
getEvent()?.addWarning(
`withings sleep: rollup recompute failed for ${userId}: ${err}`,
);
}
await recordSyncSuccess(userId, "withings");
return imported;
}
+33
View File
@@ -18,6 +18,10 @@ import {
recordSyncFailure,
recordSyncSuccess,
} from "@/lib/integrations/status";
import {
collapseToTypeDayKeys,
recomputeBucketsForMeasurement,
} from "@/lib/measurements/rollups";
/**
* Build the callback URL handed to Withings at `Notify.subscribe` time.
@@ -194,6 +198,18 @@ export async function syncUserMeasurements(
}
let imported = 0;
// v1.4.39.1 — track every (type, measuredAt) we touched so the
// persistent rollup tier can be re-folded at the end of the sync.
// Pre-fix, Withings-ingested BP / weight / pulse / body-fat rows
// skipped the rollup write hook entirely: subsequent dashboard chart
// fetches with `source=rollup` saw fewer DAY buckets than the live
// measurements table, painting the "Noch nicht genug Daten" empty
// state for the 30-day range on accounts whose recent data only came
// through Withings. We collect first, fold once at the end so a 100-
// row monthly catch-up costs at most ~N (type, day) recomputes rather
// than N per-row hooks. Best-effort: a populator hiccup never fails
// the user's sync.
const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
for (const m of measures) {
const measType = m.type as MeasurementType;
@@ -233,12 +249,29 @@ export async function syncUserMeasurements(
},
});
}
touched.push({ type: measType, measuredAt: m.measuredAt });
imported++;
} catch (err) {
getEvent()?.addWarning(`Failed to upsert measure: ${err}`);
}
}
// 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);
for (const k of keys) {
await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
}
} catch (err) {
getEvent()?.addWarning(
`withings: rollup recompute failed for ${userId}: ${err}`,
);
}
// Update last synced timestamp
await prisma.withingsConnection.update({
where: { userId },