Bound the sleep-timeline delete to the window the re-sync can restore

The Withings arm of the backfill deleted the source's entire
SLEEP_DURATION history while its re-sync fetches only the trailing 30
days regardless of fullSync — every older night was raw data loss with
orphaned rollup buckets on top. The delete is now bounded to the
re-sync window and the touched rollup span is recomputed afterwards;
WHOOP keeps the whole-history clear because its full sync genuinely
re-imports every night.
This commit is contained in:
Marc-André Bombeck committed 2026-08-14 02:29:01 +02:00
1 parent 2c0b4d60dd
commit b49013e6c6
3 files changed
+107 -22

No files matched your search

@@ -9,18 +9,23 @@
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const { prismaMock, bossSend, syncUserWhoop, syncWithingsSleep } = vi.hoisted(
() => ({
prismaMock: {
whoopConnection: { findMany: vi.fn(), update: vi.fn() },
withingsConnection: { findMany: vi.fn(), update: vi.fn() },
measurement: { deleteMany: vi.fn() },
},
bossSend: vi.fn(),
syncUserWhoop: vi.fn(),
syncWithingsSleep: vi.fn(),
}),
);
const {
prismaMock,
bossSend,
syncUserWhoop,
syncWithingsSleep,
recomputeRollups,
} = vi.hoisted(() => ({
prismaMock: {
whoopConnection: { findMany: vi.fn(), update: vi.fn() },
withingsConnection: { findMany: vi.fn(), update: vi.fn() },
measurement: { deleteMany: vi.fn() },
},
bossSend: vi.fn(),
syncUserWhoop: vi.fn(),
syncWithingsSleep: vi.fn(),
recomputeRollups: vi.fn(),
}));
vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
@@ -33,9 +38,14 @@ vi.mock("@/lib/whoop/sync", () => ({
}));
vi.mock("@/lib/withings/sync-sleep", () => ({
SLEEP_BACKFILL_DAYS: 30,
syncUserSleep: (...a: unknown[]) => syncWithingsSleep(...a),
}));
vi.mock("@/lib/rollups/measurement-rollups", () => ({
recomputeUserRollups: (...a: unknown[]) => recomputeRollups(...a),
}));
vi.mock("@/lib/logging/context", () => ({
annotate: () => {},
getEvent: () => null,
@@ -138,10 +148,16 @@ describe("runSleepTimelineBackfillForUser", () => {
expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled();
});
it("WITHINGS: deletes the source's sleep rows, re-syncs, and stamps the marker", async () => {
// Watched red: with the unbounded delete restored (no `measuredAt` bound)
// this fails on the where-clause assertion — the pre-fix pass deleted the
// whole Withings sleep history while the re-sync restored only the
// trailing 30-day window, permanent raw data loss beyond it.
it("WITHINGS: deletes ONLY the re-sync window, re-syncs, recomputes the span, and stamps the marker", async () => {
prismaMock.measurement.deleteMany.mockResolvedValue({ count: 3 });
syncWithingsSleep.mockResolvedValue(7);
recomputeRollups.mockResolvedValue(undefined);
prismaMock.withingsConnection.update.mockResolvedValue({});
const before = Date.now();
const { deleted, imported } = await runSleepTimelineBackfillForUser(
"v1",
@@ -150,12 +166,46 @@ describe("runSleepTimelineBackfillForUser", () => {
expect(deleted).toBe(3);
expect(imported).toBe(7);
expect(prismaMock.measurement.deleteMany).toHaveBeenCalledWith({
where: { userId: "v1", type: "SLEEP_DURATION", source: "WITHINGS" },
});
const deleteArg = prismaMock.measurement.deleteMany.mock.calls[0]![0] as {
where: {
userId: string;
type: string;
source: string;
measuredAt: { gte: Date };
};
};
expect(deleteArg.where.userId).toBe("v1");
expect(deleteArg.where.type).toBe("SLEEP_DURATION");
expect(deleteArg.where.source).toBe("WITHINGS");
// The bound matches the 30-day window the re-sync actually covers.
const windowMs = 30 * 24 * 60 * 60 * 1000;
const gte = deleteArg.where.measuredAt.gte.getTime();
expect(Math.abs(before - windowMs - gte)).toBeLessThan(5_000);
expect(syncWithingsSleep).toHaveBeenCalledWith("v1", { fullSync: true });
// The rollup span the delete touched is recomputed, so a day whose
// segments Withings no longer returns cannot keep an orphan bucket.
const rollupArg = recomputeRollups.mock.calls[0] as unknown as [
string,
{ types: string[]; from: Date; to: Date },
];
expect(rollupArg[0]).toBe("v1");
expect(rollupArg[1].types).toEqual(["SLEEP_DURATION"]);
const updateArg = prismaMock.withingsConnection.update.mock.calls[0]![0];
expect(updateArg.where).toEqual({ userId: "v1" });
expect(updateArg.data.sleepTimelineBackfillAt).toBeInstanceOf(Date);
});
it("WHOOP: still clears the whole history — its full sync re-imports every night", async () => {
prismaMock.measurement.deleteMany.mockResolvedValue({ count: 9 });
syncUserWhoop.mockResolvedValue({ imported: 12, failed: false });
prismaMock.whoopConnection.update.mockResolvedValue({});
await runSleepTimelineBackfillForUser("w1", "WHOOP");
expect(prismaMock.measurement.deleteMany).toHaveBeenCalledWith({
where: { userId: "w1", type: "SLEEP_DURATION", source: "WHOOP" },
});
});
});
+37 -5
View File
@@ -35,7 +35,11 @@ import { annotate } from "@/lib/logging/context";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
import { syncUserWhoop } from "@/lib/whoop/sync";
import { syncUserSleep as syncWithingsSleep } from "@/lib/withings/sync-sleep";
import {
SLEEP_BACKFILL_DAYS,
syncUserSleep as syncWithingsSleep,
} from "@/lib/withings/sync-sleep";
import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups";
export const SLEEP_TIMELINE_BACKFILL_QUEUE = "sleep-timeline-backfill";
@@ -65,12 +69,15 @@ export async function runSleepTimelineBackfillForUser(
userId: string,
provider: SleepTimelineProvider,
): Promise<{ deleted: number; imported: number }> {
const { count: deleted } = await prisma.measurement.deleteMany({
where: { userId, type: "SLEEP_DURATION", source: provider },
});
let deleted = 0;
let imported = 0;
if (provider === "WHOOP") {
// WHOOP's `fullSync` walks the whole collection from the far-past
// anchor, so the delete may cover the whole history — every night is
// re-imported with the corrected segment timeline.
({ count: deleted } = await prisma.measurement.deleteMany({
where: { userId, type: "SLEEP_DURATION", source: provider },
}));
// `syncUserWhoop` now returns a verdict; gate the completion marker on a
// clean run so a partial re-sync (dead token, write failure) throws for a
// pg-boss retry instead of freezing the timeline gap under a stamped
@@ -87,7 +94,32 @@ export async function runSleepTimelineBackfillForUser(
data: { sleepTimelineBackfillAt: new Date() },
});
} else {
// Withings re-syncs ONLY the trailing `SLEEP_BACKFILL_DAYS` window
// (`sync-sleep.ts` fetches that span regardless of `fullSync`), so the
// delete is bounded to the same window. The unbounded delete this
// replaces threw away every older night's raw rows — data the re-sync
// could never restore — and left their rollup buckets orphaned.
const windowStart = new Date(
Date.now() - SLEEP_BACKFILL_DAYS * 24 * 60 * 60 * 1000,
);
({ count: deleted } = await prisma.measurement.deleteMany({
where: {
userId,
type: "SLEEP_DURATION",
source: provider,
measuredAt: { gte: windowStart },
},
}));
imported = await syncWithingsSleep(userId, { fullSync: true });
// Recompute the rollup span the delete touched. The re-sync tail
// re-folds only the days it re-imported; a day inside the window whose
// segments Withings no longer returns would otherwise keep a bucket
// over rows that are gone.
await recomputeUserRollups(userId, {
types: ["SLEEP_DURATION"],
from: windowStart,
to: new Date(),
});
await prisma.withingsConnection.update({
where: { userId },
data: { sleepTimelineBackfillAt: new Date() },
+4 -1
View File
@@ -164,7 +164,10 @@ export function mapWithingsSleepSummary(
* Matches every other Withings ingest path so a reconnect lights up
* 30 days of stage data.
*/
const SLEEP_BACKFILL_DAYS = 30;
// Exported so the sleep-timeline backfill can bound its DELETE to exactly
// the window this sync will re-import — deleting beyond it is raw data loss
// the re-sync can never restore.
export const SLEEP_BACKFILL_DAYS = 30;
/**
* One stage segment from the `action=get` series. Withings emits