test(fitbit): cover the stable sleep key, the sweep, and the one-shot repair

Mirrors the Google sleep suites name for name: the mapper keeps its externalIds
across a re-score that re-classifies and renumbers segments, the sweep stays
bounded to the night's window and never touches the fresh keep-set, a null
window issues no delete at all, and a cleanup failure never fails the sync.

The end-to-end fixture runs one night stored under the old keys through
`syncUserSleep` and pins the ordering the heal depends on — sweep first, then
the rescue re-keying the tombstoned rows in place, no parallel insert — plus a
second pass that writes nothing at all.

Also covers the repair's verdict gates (bounded 365-day window, marker stamped
only on a clean run, watermark untouched, dead token and reauth park returning
without a stamp), the unchanged-value skip including the re-classified segment
it must not swallow, and both Retry-After paths.
This commit is contained in:
Marc-André Bombeck committed 2026-07-25 06:15:37 +02:00
1 parent ce41d5bf17
commit a08c759763
7 files changed
+953 -11

No files matched your search

+236 -10
View File
@@ -2,6 +2,7 @@ import { createHash } from "node:crypto";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import {
FITBIT_API_BASE,
FITBIT_RETRY_AFTER_WAIT_CAP_S,
FITBIT_FIELD_MAP,
FITBIT_OAUTH_SCOPE,
exchangeCode,
@@ -20,7 +21,7 @@ import {
mapOxygenSaturation,
mapRespiratoryRate,
mapRestingHeartRate,
mapSleepSession,
mapSleepSessionDetailed,
mapSteps,
mapVo2Max,
mapWeight,
@@ -35,13 +36,20 @@ import {
const CREDS = { clientId: "cid", clientSecret: "csecret" };
/** Stub global fetch with a queue of `{ status, body }` responses. */
function installFetchMock(pages: Array<{ status: number; body: unknown }>) {
function installFetchMock(
pages: Array<{
status: number;
body: unknown;
headers?: Record<string, string>;
}>,
) {
let i = 0;
const fetchMock = vi.fn(async () => {
const page = pages[Math.min(i, pages.length - 1)]!;
i += 1;
return {
status: page.status,
headers: new Headers(page.headers ?? {}),
json: async () => page.body,
};
});
@@ -487,7 +495,8 @@ describe("sleep mapping (1.2 levels.data)", () => {
],
});
expect(sessions).toHaveLength(1);
const out = mapSleepSession(sessions[0]!);
const mapped = mapSleepSessionDetailed(sessions[0]!);
const out = mapped.rows;
// One row per segment.
expect(out).toHaveLength(3);
@@ -505,15 +514,123 @@ describe("sleep mapping (1.2 levels.data)", () => {
const rem = out.find((m) => m.sleepStage === "REM")!;
expect(rem.value).toBe(45);
// Every fieldTag distinct (logId anchor + stage + segment index).
// Every fieldTag distinct: the logId anchor plus the segment's own START
// instant — no positional index, no stage label.
const tags = out.map((m) => m.fieldTag);
expect(new Set(tags).size).toBe(tags.length);
expect(deep.fieldTag).toBe("999:sleep_deep:1");
expect(deep.fieldTag).toBe(
`999:sleep:${new Date("2026-05-10T22:30:00.000").toISOString()}`,
);
// The session window spans the earliest segment start → the latest end.
expect(mapped.windowStart!.getTime()).toBe(
new Date("2026-05-10T22:00:00.000").getTime(),
);
expect(mapped.windowEnd!.getTime()).toBe(
new Date("2026-05-11T00:15:00.000").getTime(),
);
});
it("keeps stable per-segment externalIds across a re-score (anchor = logId)", () => {
// Two fetches of the SAME night. On the re-score Fitbit RE-CLASSIFIES the
// first block (light→deep), SPLITS the second, and thereby renumbers every
// following segment — the old positional, stage-scoped key minted fresh
// externalIds for unchanged blocks, so the upsert inserted parallel rows
// the night total then double-counted. The anchor + segment-start key must
// yield the SAME externalIds for the blocks that did not move.
const night = (
data: Array<{ dateTime: string; level: string; seconds: number }>,
) =>
mapSleepSessionDetailed({
logId: 777,
startTime: "2026-05-10T22:00:00.000",
endTime: "2026-05-11T06:00:00.000",
levels: { data },
});
const first = night([
{ dateTime: "2026-05-10T22:00:00.000", level: "light", seconds: 1800 },
{ dateTime: "2026-05-10T22:30:00.000", level: "rem", seconds: 3600 },
{ dateTime: "2026-05-10T23:30:00.000", level: "deep", seconds: 1800 },
]);
const rescored = night([
// Same start, re-classified → same key, stage updates in place.
{ dateTime: "2026-05-10T22:00:00.000", level: "deep", seconds: 1800 },
// Split into two halves: the first keeps the block's start key, the
// second is a genuinely new segment. Everything after it renumbers.
{ dateTime: "2026-05-10T22:30:00.000", level: "rem", seconds: 1800 },
{ dateTime: "2026-05-10T23:00:00.000", level: "light", seconds: 1800 },
{ dateTime: "2026-05-10T23:30:00.000", level: "deep", seconds: 1800 },
]);
const startKey = (iso: string) =>
`777:sleep:${new Date(iso).toISOString()}`;
expect(first.rows.map((r) => r.fieldTag)).toEqual([
startKey("2026-05-10T22:00:00.000"),
startKey("2026-05-10T22:30:00.000"),
startKey("2026-05-10T23:30:00.000"),
]);
// The unchanged blocks keep their keys across the re-score — overwrite,
// never a parallel duplicate. Only the genuinely new split half is new.
expect(rescored.rows.map((r) => r.fieldTag)).toEqual([
startKey("2026-05-10T22:00:00.000"),
startKey("2026-05-10T22:30:00.000"),
startKey("2026-05-10T23:00:00.000"),
startKey("2026-05-10T23:30:00.000"),
]);
// The re-classification rides the stage axis (an in-place update).
expect(first.rows[0]!.sleepStage).toBe("CORE"); // light → shallow-NREM band
expect(rescored.rows[0]!.sleepStage).toBe("DEEP");
});
it("keys a classic (asleep / restless / in_bed) log identically", () => {
const mapped = mapSleepSessionDetailed({
logId: 555,
levels: {
data: [
{
dateTime: "2026-05-10T22:00:00.000",
level: "asleep",
seconds: 600,
},
{
dateTime: "2026-05-10T22:10:00.000",
level: "restless",
seconds: 300,
},
{
dateTime: "2026-05-10T22:15:00.000",
level: "in_bed",
seconds: 300,
},
// Unknown label — skipped, and it must not shift the other keys.
{
dateTime: "2026-05-10T22:20:00.000",
level: "snoring",
seconds: 300,
},
],
},
});
expect(mapped.rows.map((r) => r.fieldTag)).toEqual([
`555:sleep:${new Date("2026-05-10T22:00:00.000").toISOString()}`,
`555:sleep:${new Date("2026-05-10T22:10:00.000").toISOString()}`,
`555:sleep:${new Date("2026-05-10T22:15:00.000").toISOString()}`,
]);
expect(mapped.rows.map((r) => r.sleepStage)).toEqual([
"ASLEEP",
"AWAKE",
"IN_BED",
]);
});
it("yields nothing for a session with no parseable segments", () => {
expect(mapSleepSession({ levels: { data: [] } })).toHaveLength(0);
expect(mapSleepSession({})).toHaveLength(0);
const empty = mapSleepSessionDetailed({ levels: { data: [] } });
expect(empty.rows).toHaveLength(0);
// A null window is what stops the sweep from issuing an unbounded delete.
expect(empty.windowStart).toBeNull();
expect(empty.windowEnd).toBeNull();
expect(mapSleepSessionDetailed({}).rows).toHaveLength(0);
});
it("anchors a near-midnight segment END to the user's timezone, not the process zone", () => {
@@ -540,7 +657,7 @@ describe("sleep mapping (1.2 levels.data)", () => {
],
});
const out = mapSleepSession(sessions[0]!, "Europe/Berlin");
const out = mapSleepSessionDetailed(sessions[0]!, "Europe/Berlin").rows;
expect(out).toHaveLength(1);
// 00:30 CEST (UTC+2) → 22:30 UTC on the PRIOR civil day.
expect(out[0]!.measuredAt.toISOString()).toBe("2026-05-10T22:30:00.000Z");
@@ -569,10 +686,119 @@ describe("sleep mapping (1.2 levels.data)", () => {
],
});
const out = mapSleepSession(sessions[0]!, "Europe/Berlin");
const out = mapSleepSessionDetailed(sessions[0]!, "Europe/Berlin").rows;
expect(out).toHaveLength(1);
// anchor = start instant ISO (UTC) → fieldTag starts with that instant.
expect(out[0]!.fieldTag).toBe("2026-05-10T21:00:00.000Z:sleep_rem:0");
expect(out[0]!.fieldTag).toBe(
"2026-05-10T21:00:00.000Z:sleep:2026-05-10T22:00:00.000Z",
);
});
});
describe("fitbitGet — Retry-After on a 429", () => {
const S = new Date("2026-05-01T00:00:00.000Z");
const E = new Date("2026-05-10T00:00:00.000Z");
it("waits out a Retry-After inside the cap and retries the request once", async () => {
vi.useFakeTimers({ toFake: ["setTimeout"] });
try {
const fetchMock = installFetchMock([
{ status: 429, headers: { "retry-after": "5" }, body: {} },
{ status: 200, body: { sleep: [] } },
]);
const pending = fetchSleepRange("tok", S, E);
// Still parked on the delay the upstream asked for — the retry must not
// fire early (an immediate re-request would just burn the next 429).
await vi.advanceTimersByTimeAsync(4_999);
expect(fetchMock).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
await expect(pending).resolves.toEqual({ sleep: [] });
expect(fetchMock).toHaveBeenCalledTimes(2);
} finally {
vi.useRealTimers();
}
});
it("fails fast without waiting when the delay exceeds the cap", async () => {
// An exhausted hourly budget asks for the better part of an hour. Sitting
// that out inside a worker slot starves every other user in the tick (and
// a pg-boss job would be lapped by its own expiration), so the request
// throws immediately and the next scheduled tick refetches the window.
const fetchMock = installFetchMock([
{
status: 429,
headers: { "retry-after": String(FITBIT_RETRY_AFTER_WAIT_CAP_S + 1) },
body: {},
},
{ status: 200, body: { sleep: [] } },
]);
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
name: "FitbitApiError",
classification: "transient",
httpStatus: 429,
retryAfterSeconds: FITBIT_RETRY_AFTER_WAIT_CAP_S + 1,
});
// No second attempt — and no wait.
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("throws without retrying when the header is missing", async () => {
const fetchMock = installFetchMock([
{ status: 429, body: {} },
{ status: 200, body: { sleep: [] } },
]);
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
name: "FitbitApiError",
httpStatus: 429,
retryAfterSeconds: undefined,
});
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("ignores the HTTP-date form of Retry-After (unknown wait → fail fast)", async () => {
const fetchMock = installFetchMock([
{
status: 429,
headers: { "retry-after": "Wed, 21 Oct 2026 07:28:00 GMT" },
body: {},
},
{ status: 200, body: { sleep: [] } },
]);
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
httpStatus: 429,
});
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("gives up after a second 429 (one retry, never a loop)", async () => {
const fetchMock = installFetchMock([
{ status: 429, headers: { "retry-after": "0" }, body: {} },
{ status: 429, headers: { "retry-after": "0" }, body: {} },
{ status: 200, body: { sleep: [] } },
]);
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
httpStatus: 429,
});
expect(fetchMock).toHaveBeenCalledTimes(2);
});
it("does not retry a non-429 failure", async () => {
const fetchMock = installFetchMock([
{ status: 500, body: {} },
{ status: 200, body: { sleep: [] } },
]);
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
httpStatus: 500,
});
expect(fetchMock).toHaveBeenCalledTimes(1);
});
});
@@ -0,0 +1,341 @@
/**
* Pins the safety contract of `replaceStaleFitbitSleep` — the replace-by-window
* cleanup that keeps a re-scored Fitbit night from double-counting — and the
* end-to-end heal it produces together with the stable segment key and the
* natural-key rescue.
*
* The query MUST be tightly bounded: only LIVE `FITBIT` `SLEEP_DURATION` rows,
* only inside the session window, and NEVER a row in the fresh keep-set. A
* session with no window is skipped entirely (nothing to clean, and an
* unbounded delete would be data loss).
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const {
findManyMock,
createManyMock,
updateMock,
updateManyMock,
connectionFindUniqueMock,
fetchSleepRangeMock,
callOrder,
} = vi.hoisted(() => ({
findManyMock: vi.fn(async () => [] as unknown[]),
createManyMock: vi.fn<
() => Promise<Array<{ id: string; type: string; measuredAt: Date }>>
>(async () => []),
updateMock: vi.fn(async () => ({})),
updateManyMock: vi.fn(async () => ({ count: 0 })),
connectionFindUniqueMock: vi.fn(async () => null as unknown),
fetchSleepRangeMock: vi.fn(async () => ({}) as unknown),
callOrder: [] as string[],
}));
vi.mock("@/lib/db", () => ({
prisma: {
measurement: {
findMany: findManyMock,
createManyAndReturn: createManyMock,
update: updateMock,
updateMany: updateManyMock,
},
fitbitConnection: { findUnique: connectionFindUniqueMock },
},
}));
vi.mock("@/lib/crypto", () => ({
encrypt: vi.fn((v: string) => v),
decrypt: vi.fn((v: string) => v),
}));
vi.mock("@/lib/logging/context", () => ({
getEvent: () => ({ addWarning: vi.fn() }),
annotate: () => {},
}));
vi.mock("@/lib/integrations/status", () => ({
isReauthRequired: vi.fn(async () => false),
recordSyncFailure: vi.fn(async () => {}),
recordSyncSuccess: vi.fn(async () => {}),
}));
vi.mock("@/lib/rollups/measurement-rollups", () => ({
collapseToTypeDayKeys: vi.fn(() => []),
recomputeBucketsForMeasurement: vi.fn(async () => {}),
recomputeUserRollups: vi.fn(async () => {}),
}));
vi.mock("@/lib/insights/comprehensive-generate", () => ({
invalidateStatusInsightsForTypes: vi.fn(async () => {}),
}));
vi.mock("@/lib/integrations/oauth-refresh", () => ({
persistRotatedToken: vi.fn(async () => {}),
}));
vi.mock("@/lib/arrivals/measurement-emit", () => ({
emitInsertedMeasurementArrivals: vi.fn(async () => {}),
}));
vi.mock("@/lib/daily/morning-refresh-trigger", () => ({
maybeEnqueueMorningRefresh: vi.fn(async () => {}),
}));
vi.mock("@/lib/tz/resolver", () => ({
resolveUserTimezone: vi.fn(async () => "UTC"),
}));
vi.mock("../credentials", () => ({
getUserFitbitCredentials: vi.fn(async () => null),
}));
vi.mock("../client", async () => {
const actual = await vi.importActual<typeof import("../client")>("../client");
return { ...actual, fetchSleepRange: fetchSleepRangeMock };
});
import { replaceStaleFitbitSleep } from "../sync-core";
import { syncUserSleep } from "../sync-sleep";
beforeEach(() => {
vi.clearAllMocks();
callOrder.length = 0;
findManyMock.mockReset().mockImplementation(async () => {
callOrder.push("probe");
return [];
});
createManyMock.mockReset().mockImplementation(async () => {
callOrder.push("create");
return [];
});
updateMock.mockReset().mockImplementation(async () => {
callOrder.push("update");
return {};
});
updateManyMock.mockReset().mockImplementation(async () => {
callOrder.push("sweep");
return { count: 0 };
});
});
describe("replaceStaleFitbitSleep — bounded replace-by-window", () => {
it("soft-deletes only live FITBIT sleep rows in the window, excluding the fresh set", async () => {
updateManyMock.mockImplementation(async () => {
callOrder.push("sweep");
return { count: 1 };
});
const windowStart = new Date("2026-06-01T22:30:00.000Z");
const windowEnd = new Date("2026-06-02T06:30:00.000Z");
const keepIds = ["777:sleep:a", "777:sleep:b"];
const removed = await replaceStaleFitbitSleep("user-1", [
{ windowStart, windowEnd, keepIds },
]);
expect(removed).toBe(1);
expect(updateManyMock).toHaveBeenCalledTimes(1);
const arg = (updateManyMock.mock.calls[0]! as unknown[])[0] as {
where: Record<string, unknown>;
data: Record<string, unknown>;
};
expect(arg.where).toEqual({
userId: "user-1",
source: "FITBIT",
type: "SLEEP_DURATION",
deletedAt: null,
measuredAt: { gte: windowStart, lte: windowEnd },
externalId: { notIn: keepIds },
});
// A soft delete — the row is tombstoned, never hard-removed.
expect(arg.data).toHaveProperty("deletedAt");
expect(arg.data.deletedAt).toBeInstanceOf(Date);
});
it("skips a session with a null window (never issues an unbounded delete)", async () => {
await replaceStaleFitbitSleep("user-1", [
{ windowStart: null, windowEnd: null, keepIds: ["x"] },
{
windowStart: new Date("2026-06-01T22:00:00.000Z"),
windowEnd: new Date("2026-06-02T06:00:00.000Z"),
keepIds: [],
},
]);
// Both sessions are unsafe to clean (no window, or no fresh ids to keep) —
// neither may issue a delete.
expect(updateManyMock).not.toHaveBeenCalled();
});
it("never fails the sync when the cleanup query throws", async () => {
updateManyMock.mockRejectedValueOnce(new Error("db down"));
await expect(
replaceStaleFitbitSleep("user-1", [
{
windowStart: new Date("2026-06-01T22:00:00.000Z"),
windowEnd: new Date("2026-06-02T06:00:00.000Z"),
keepIds: ["777:sleep:a"],
},
]),
).resolves.toBe(0);
});
});
// ── End-to-end: a re-scored night heals through the write path ──────────────
//
// One night stored under the OLD key format, then re-fetched after a re-score.
// The sweep tombstones the old-keyed rows FIRST, so the upsert's natural-key
// rescue re-keys those very rows in place — no parallel copy, no silent drop
// against the natural unique index, and the night total stays the fresh total.
const SEG_A_START = "2026-06-01T22:00:00.000Z";
const SEG_A_END = new Date("2026-06-01T22:30:00.000Z");
const SEG_B_START = "2026-06-01T22:30:00.000Z";
const SEG_B_END = new Date("2026-06-01T23:30:00.000Z");
const NEW_KEY_A = `777:sleep:${SEG_A_START}`;
const NEW_KEY_B = `777:sleep:${SEG_B_START}`;
const RESCORED_NIGHT = {
sleep: [
{
logId: 777,
startTime: SEG_A_START,
endTime: SEG_B_END.toISOString(),
levels: {
data: [
// Re-classified from light → deep on the SAME block.
{ dateTime: SEG_A_START, level: "deep", seconds: 1800 },
{ dateTime: SEG_B_START, level: "rem", seconds: 3600 },
],
},
},
],
};
const SYNC_WINDOW = {
start: new Date("2026-06-01T00:00:00.000Z"),
end: new Date("2026-06-03T00:00:00.000Z"),
deferRollup: true,
};
function connectionAlive() {
connectionFindUniqueMock.mockResolvedValue({
id: "conn-1",
fitbitUserId: "fb-1",
accessToken: "tok",
refreshToken: "ref",
tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
});
}
describe("syncUserSleep — a re-scored night neither duplicates nor disappears", () => {
beforeEach(() => {
connectionAlive();
fetchSleepRangeMock.mockResolvedValue(RESCORED_NIGHT);
});
it("sweeps the night's window BEFORE the upsert and re-keys the tombstoned rows in place", async () => {
updateManyMock.mockImplementation(async () => {
callOrder.push("sweep");
return { count: 2 };
});
findManyMock
.mockImplementationOnce(async () => {
// externalId probe: the fresh keys do not exist yet (the stored rows
// still carry the old positional format).
callOrder.push("probe:externalId");
return [];
})
.mockImplementationOnce(async () => {
// natural-key probe: the rows the sweep just tombstoned occupy the
// same (type, measuredAt, sleepStage) slots.
callOrder.push("probe:naturalKey");
return [
{
id: "old-a",
type: "SLEEP_DURATION",
measuredAt: SEG_A_END,
sleepStage: "DEEP",
},
{
id: "old-b",
type: "SLEEP_DURATION",
measuredAt: SEG_B_END,
sleepStage: "REM",
},
];
});
const imported = await syncUserSleep("user-1", SYNC_WINDOW);
// The sweep runs first — reversing the order would tombstone the rows the
// upsert had just written.
expect(callOrder[0]).toBe("sweep");
expect(callOrder.indexOf("sweep")).toBeLessThan(
callOrder.indexOf("probe:externalId"),
);
const sweepArg = (updateManyMock.mock.calls[0]! as unknown[])[0] as {
where: Record<string, unknown>;
};
expect(sweepArg.where).toMatchObject({
userId: "user-1",
source: "FITBIT",
type: "SLEEP_DURATION",
deletedAt: null,
measuredAt: {
gte: new Date(SEG_A_START),
lte: SEG_B_END,
},
externalId: { notIn: [NEW_KEY_A, NEW_KEY_B] },
});
// No parallel copy: both rows are re-keyed in place, none inserted.
expect(createManyMock).not.toHaveBeenCalled();
expect(updateMock).toHaveBeenCalledTimes(2);
const updates = updateMock.mock.calls.map(
(c) =>
(c as unknown[])[0] as {
where: { id: string };
data: Record<string, unknown>;
},
);
expect(updates.map((u) => u.where.id)).toEqual(["old-a", "old-b"]);
expect(updates.map((u) => u.data.externalId)).toEqual([
NEW_KEY_A,
NEW_KEY_B,
]);
for (const u of updates) {
expect(u.data.deletedAt).toBeNull();
}
// The re-classification lands on the stage axis of the same row.
expect(updates[0]!.data.sleepStage).toBe("DEEP");
// Two segments in, two rows out — the night total is the fresh total.
expect(imported).toBe(2);
});
it("is a no-op on the second pass (idempotent — no churn, no duplicates)", async () => {
findManyMock.mockImplementation(async () => {
callOrder.push("probe");
// Both rows now carry the fresh keys and the fresh payload.
return [
{
id: "row-a",
type: "SLEEP_DURATION",
externalId: NEW_KEY_A,
value: 30,
unit: "minutes",
measuredAt: SEG_A_END,
sleepStage: "DEEP",
deletedAt: null,
},
{
id: "row-b",
type: "SLEEP_DURATION",
externalId: NEW_KEY_B,
value: 60,
unit: "minutes",
measuredAt: SEG_B_END,
sleepStage: "REM",
deletedAt: null,
},
];
});
const imported = await syncUserSleep("user-1", SYNC_WINDOW);
// The sweep still runs, but it excludes both fresh keys, so it removes
// nothing; the upsert writes nothing at all.
expect(updateManyMock).toHaveBeenCalledTimes(1);
expect(createManyMock).not.toHaveBeenCalled();
expect(updateMock).not.toHaveBeenCalled();
expect(imported).toBe(0);
});
});
@@ -98,6 +98,11 @@ describe("upsertFitbitMeasurements — tombstones resurrect", () => {
id: "row-1",
type: "ACTIVITY_STEPS",
externalId: "stats:steps:2026-07-08",
value: 8123,
unit: "steps",
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
sleepStage: null,
deletedAt: new Date("2026-07-09T00:00:00.000Z"),
},
]);
@@ -136,6 +141,127 @@ describe("upsertFitbitMeasurements — tombstones resurrect", () => {
});
});
describe("upsertFitbitMeasurements — no-op overwrite skip", () => {
it("skips the update entirely for an identical LIVE row (no syncVersion churn)", async () => {
findManyMock.mockResolvedValue([
{
id: "row-1",
type: "ACTIVITY_STEPS",
externalId: "stats:steps:2026-07-08",
value: 8123,
unit: "steps",
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
sleepStage: null,
deletedAt: null,
},
]);
const { imported } = await upsertFitbitMeasurements(
"user-1",
[STEPS_READING],
{ deferRollup: true },
);
// The 24 h overlap re-fetches this row hourly; an unchanged live row must
// not be rewritten (the unconditional syncVersion bump churned the DB and
// every iOS delta pull).
expect(updateMock).not.toHaveBeenCalled();
expect(createManyMock).not.toHaveBeenCalled();
expect(imported).toBe(0);
});
it("still writes when any payload field differs on a live row", async () => {
findManyMock.mockResolvedValue([
{
id: "row-1",
type: "ACTIVITY_STEPS",
externalId: "stats:steps:2026-07-08",
value: 7999, // stale daily total — Fitbit re-rolled the day upward
unit: "steps",
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
sleepStage: null,
deletedAt: null,
},
]);
const { imported } = await upsertFitbitMeasurements(
"user-1",
[STEPS_READING],
{ deferRollup: true },
);
expect(updateMock).toHaveBeenCalledTimes(1);
expect(imported).toBe(1);
});
it("writes when only the sleep stage differs (a re-classified segment)", async () => {
// The stable segment key deliberately drops the stage, so a re-label
// arrives on the SAME externalId — the skip must not swallow it.
findManyMock.mockResolvedValue([
{
id: "row-1",
type: "SLEEP_DURATION",
externalId: "777:sleep:2026-07-08T05:45:00.000Z",
value: 45,
unit: "minutes",
measuredAt: new Date("2026-07-08T06:30:00.000Z"),
sleepStage: "CORE",
deletedAt: null,
},
]);
const { imported } = await upsertFitbitMeasurements(
"user-1",
[
{
type: "SLEEP_DURATION",
value: 45,
unit: "minutes",
measuredAt: new Date("2026-07-08T06:30:00.000Z"),
externalId: "777:sleep:2026-07-08T05:45:00.000Z",
sleepStage: "DEEP" as const,
},
],
{ deferRollup: true },
);
expect(updateMock).toHaveBeenCalledTimes(1);
const upd = (updateMock.mock.calls[0] as unknown[])[0] as {
data: Record<string, unknown>;
};
expect(upd.data.sleepStage).toBe("DEEP");
expect(imported).toBe(1);
});
it("an identical TOMBSTONED row ALWAYS updates — the write is the resurrection", async () => {
findManyMock.mockResolvedValue([
{
id: "row-1",
type: "ACTIVITY_STEPS",
externalId: "stats:steps:2026-07-08",
value: 8123,
unit: "steps",
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
sleepStage: null,
deletedAt: new Date("2026-07-09T00:00:00.000Z"),
},
]);
const { imported } = await upsertFitbitMeasurements(
"user-1",
[STEPS_READING],
{ deferRollup: true },
);
expect(updateMock).toHaveBeenCalledTimes(1);
const upd = (updateMock.mock.calls[0] as unknown[])[0] as {
data: Record<string, unknown>;
};
expect(upd.data.deletedAt).toBeNull();
expect(imported).toBe(1);
});
});
describe("upsertFitbitMeasurements — natural-key migration rescue", () => {
const SLEEP_READING = {
type: "SLEEP_DURATION",
@@ -34,6 +34,7 @@ const source =
const FITBIT_QUEUE_CONSTS = [
"FITBIT_SYNC_QUEUE",
"FITBIT_BACKFILL_QUEUE",
"FITBIT_SLEEP_REPAIR_QUEUE",
"FITBIT_OAUTH_STATE_CLEANUP_QUEUE",
] as const;
@@ -68,6 +69,9 @@ describe("reminder-worker — Fitbit sync queues", () => {
expect(source).toMatch(
/boss\.work[\s\S]{0,200}FITBIT_BACKFILL_QUEUE[\s\S]{0,260}enqueueIntegrationBackfillAdmission/,
);
expect(source).toMatch(
/boss\.work[\s\S]{0,200}FITBIT_SLEEP_REPAIR_QUEUE[\s\S]{0,260}enqueueIntegrationBackfillAdmission/,
);
expect(source).toMatch(
/boss\.work[\s\S]{0,160}FITBIT_OAUTH_STATE_CLEANUP_QUEUE[\s\S]{0,160}handleFitbitOAuthStateCleanup/,
);
@@ -79,6 +83,13 @@ describe("reminder-worker — Fitbit sync queues", () => {
);
});
it("wires the one-shot Fitbit sleep-repair boot discovery", () => {
expect(source).toMatch(
/enqueueBootTimeFitbitSleepRepair\(\s*bootStaggerSecondsFor\("fitbit-sleep-repair"\)/,
);
expect(source).toMatch(/from\s*["']@\/lib\/jobs\/fitbit-sleep-repair["']/);
});
it("imports the Fitbit sync driver + backfill exports", () => {
expect(source).toMatch(/from\s*["']@\/lib\/fitbit\/sync["']/);
expect(source).toMatch(/from\s*["']@\/lib\/jobs\/fitbit-backfill["']/);
@@ -0,0 +1,233 @@
/**
* Fitbit one-shot sleep duplicate-repair self-convergence tests (mocked).
* - discovery only matches connections not yet repaired
* (`sleepRepairedAt IS NULL`) and singleton-keys the enqueue per user;
* - a completed pass re-reads the BOUNDED sleep history (an explicit 365-day
* `start`, not `syncUserSleep`'s 30-day default), stamps `sleepRepairedAt`
* so the next discovery drops the account, and NEVER touches the
* `lastSyncedAt` sync watermark;
* - the discovery is best-effort — errors surface through the result value.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const { prismaMock, bossSend, syncUserSleep, isReauthRequiredMock } =
vi.hoisted(() => ({
prismaMock: {
fitbitConnection: { findMany: vi.fn(), update: vi.fn() },
},
bossSend: vi.fn(),
syncUserSleep: vi.fn(),
isReauthRequiredMock: vi.fn(async () => false),
}));
vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
vi.mock("@/lib/jobs/boss-instance", () => ({
getGlobalBoss: () => ({ send: bossSend }),
}));
vi.mock("@/lib/fitbit/sync-sleep", () => ({
syncUserSleep: (...a: unknown[]) => syncUserSleep(...a),
}));
vi.mock("@/lib/logging/context", () => ({
annotate: () => {},
getEvent: () => null,
}));
vi.mock("@/lib/integrations/status", () => ({
isReauthRequired: isReauthRequiredMock,
recordSyncFailure: vi.fn(async () => {}),
recordSyncSuccess: vi.fn(async () => {}),
}));
// The repair module threads the REAL hard-fail ledger from
// `@/lib/fitbit/sync-core`; its heavy transitive deps are mocked the same way
// the Fitbit upsert suites do.
vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() }));
vi.mock("@/lib/rollups/measurement-rollups", () => ({
collapseToTypeDayKeys: vi.fn(() => []),
recomputeBucketsForMeasurement: vi.fn(async () => {}),
recomputeUserRollups: vi.fn(async () => {}),
}));
vi.mock("@/lib/insights/comprehensive-generate", () => ({
invalidateStatusInsightsForTypes: vi.fn(async () => {}),
}));
vi.mock("@/lib/integrations/oauth-refresh", () => ({
persistRotatedToken: vi.fn(async () => {}),
}));
vi.mock("@/lib/fitbit/credentials", () => ({
getUserFitbitCredentials: vi.fn(async () => null),
}));
vi.mock("@/lib/fitbit/client", () => ({ refreshAccessToken: vi.fn() }));
import {
FITBIT_BACKFILL_DAYS,
FITBIT_TOKEN_HARD_FAIL,
noteHardFailure,
noteSleepSwept,
} from "@/lib/fitbit/sync-core";
import {
enqueueBootTimeFitbitSleepRepair,
runFitbitSleepRepairForUser,
} from "../fitbit-sleep-repair";
beforeEach(() => {
vi.clearAllMocks();
isReauthRequiredMock.mockResolvedValue(false);
});
describe("enqueueBootTimeFitbitSleepRepair — discovery", () => {
it("queries only un-repaired connections and enqueues one singleton-keyed job per user", async () => {
prismaMock.fitbitConnection.findMany.mockResolvedValue([
{ userId: "u1" },
{ userId: "u2" },
]);
bossSend.mockResolvedValue("job-id");
const result = await enqueueBootTimeFitbitSleepRepair();
expect(prismaMock.fitbitConnection.findMany).toHaveBeenCalledWith(
expect.objectContaining({ where: { sleepRepairedAt: null } }),
);
expect(result.enqueued).toBe(2);
expect(bossSend).toHaveBeenCalledWith(
"fitbit-sleep-repair",
expect.objectContaining({ userId: "u1" }),
expect.objectContaining({ singletonKey: "fitbit-sleep-repair|u1" }),
);
expect(bossSend).toHaveBeenCalledWith(
"fitbit-sleep-repair",
expect.objectContaining({ userId: "u2" }),
expect.objectContaining({ singletonKey: "fitbit-sleep-repair|u2" }),
);
});
it("self-converges: no un-repaired connections → nothing enqueued", async () => {
prismaMock.fitbitConnection.findMany.mockResolvedValue([]);
const result = await enqueueBootTimeFitbitSleepRepair();
expect(result.enqueued).toBe(0);
expect(bossSend).not.toHaveBeenCalled();
});
it("never throws — surfaces a discovery error through the result value", async () => {
prismaMock.fitbitConnection.findMany.mockRejectedValue(
new Error("db down"),
);
const result = await enqueueBootTimeFitbitSleepRepair();
expect(result.error).toBe("db down");
expect(result.enqueued).toBe(0);
});
});
describe("runFitbitSleepRepairForUser", () => {
it("re-reads the bounded 365-day window (NOT the 30-day default) and stamps sleepRepairedAt", async () => {
syncUserSleep.mockResolvedValue(42);
prismaMock.fitbitConnection.update.mockResolvedValue({});
const before = Date.now();
const { imported } = await runFitbitSleepRepairForUser("u1");
const after = Date.now();
expect(imported).toBe(42);
// The explicit lower bound is what makes the repair reach history: the
// resource sync's own default only walks 30 days back.
const syncOpts = syncUserSleep.mock.calls[0]![1] as { start: Date };
expect(syncUserSleep).toHaveBeenCalledWith("u1", expect.any(Object));
expect(FITBIT_BACKFILL_DAYS).toBe(365);
const horizonMs = FITBIT_BACKFILL_DAYS * 24 * 60 * 60 * 1000;
expect(syncOpts.start).toBeInstanceOf(Date);
expect(syncOpts.start.getTime()).toBeGreaterThanOrEqual(
before - horizonMs - 1000,
);
expect(syncOpts.start.getTime()).toBeLessThanOrEqual(after - horizonMs);
const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0] as {
where: Record<string, unknown>;
data: Record<string, unknown>;
};
expect(updateArg.where).toEqual({ userId: "u1" });
expect(updateArg.data.sleepRepairedAt).toBeInstanceOf(Date);
});
it("does NOT touch the lastSyncedAt sync watermark", async () => {
syncUserSleep.mockResolvedValue(7);
prismaMock.fitbitConnection.update.mockResolvedValue({});
await runFitbitSleepRepairForUser("u1");
// The single connection write stamps ONLY the repair marker — the
// watermark stays where the incremental orchestrator left it.
for (const call of prismaMock.fitbitConnection.update.mock.calls) {
const { data } = call[0] as { data: Record<string, unknown> };
expect(data).not.toHaveProperty("lastSyncedAt");
expect(Object.keys(data)).toEqual(["sleepRepairedAt"]);
}
expect(prismaMock.fitbitConnection.update).toHaveBeenCalledTimes(1);
});
it("reports the rows the replace-by-window sweep removed (the deploy signal)", async () => {
syncUserSleep.mockImplementation(async () => {
noteSleepSwept(9);
return 12;
});
prismaMock.fitbitConnection.update.mockResolvedValue({});
await expect(runFitbitSleepRepairForUser("u1")).resolves.toEqual({
imported: 12,
removed: 9,
});
});
it("propagates a sync failure without stamping the marker (pg-boss retries)", async () => {
syncUserSleep.mockRejectedValue(new Error("fitbit 500"));
await expect(runFitbitSleepRepairForUser("u1")).rejects.toThrow(
"fitbit 500",
);
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
});
it("a swallowed hard failure (ledger entry) throws and does NOT stamp — pg-boss retries", async () => {
// `syncUserSleep` swallows fetch/write hard failures into the ambient
// ledger and still resolves a count; the repair must read the ledger, not
// the count, before stamping.
syncUserSleep.mockImplementation(async () => {
noteHardFailure("fetchSleep");
return 5;
});
await expect(runFitbitSleepRepairForUser("u1")).rejects.toThrow(
/incomplete/,
);
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
});
it("a dead token returns WITHOUT stamping and WITHOUT throwing (boot discovery re-enqueues)", async () => {
syncUserSleep.mockImplementation(async () => {
noteHardFailure(FITBIT_TOKEN_HARD_FAIL);
return 0;
});
await expect(runFitbitSleepRepairForUser("u1")).resolves.toEqual({
imported: 0,
removed: 0,
});
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
});
it("a connection parked at error_reauth returns WITHOUT running the sync or stamping", async () => {
isReauthRequiredMock.mockResolvedValue(true);
await expect(runFitbitSleepRepairForUser("u1")).resolves.toEqual({
imported: 0,
removed: 0,
});
expect(syncUserSleep).not.toHaveBeenCalled();
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
});
});
@@ -26,6 +26,7 @@ const providerSources = [
"sleep-timeline-backfill.ts",
"lab-biomarker-backfill.ts",
"strava-backfill.ts",
"fitbit-sleep-repair.ts",
].map((file) => readFileSync(join(__dirname, "..", file), "utf8"));
describe("shared full-history backfill admission", () => {
@@ -38,9 +39,10 @@ describe("shared full-history backfill admission", () => {
"sleep-timeline-backfill",
"lab-biomarker-backfill",
"strava-backfill",
"fitbit-sleep-repair",
]);
expect(INTEGRATION_BACKFILL_BOOT_ORDER.map(bootStaggerSecondsFor)).toEqual([
30, 60, 90, 120, 150, 180, 210,
30, 60, 90, 120, 150, 180, 210, 240,
]);
expect(INTEGRATION_BACKFILL_STAGGER_SECONDS).toBe(30);
@@ -57,6 +59,7 @@ describe("shared full-history backfill admission", () => {
"sleep-timeline-backfill": "enqueueBootTimeSleepTimelineBackfill",
"lab-biomarker-backfill": "enqueueBootTimeLabBiomarkerBackfill",
"strava-backfill": "enqueueBootTimeStravaBackfill",
"fitbit-sleep-repair": "enqueueBootTimeFitbitSleepRepair",
} as const;
for (const kind of INTEGRATION_BACKFILL_BOOT_ORDER) {
expect(registrarSource).toMatch(
@@ -140,6 +143,7 @@ describe("shared full-history backfill admission", () => {
"SLEEP_TIMELINE_BACKFILL_QUEUE",
"LAB_BIOMARKER_BACKFILL_QUEUE",
"STRAVA_BACKFILL_QUEUE",
"FITBIT_SLEEP_REPAIR_QUEUE",
]) {
expect(registrarSource).toMatch(
new RegExp(
@@ -92,6 +92,7 @@ describe("integration-sync registrar queue policies", () => {
["GOOGLE_HEALTH_BACKFILL_QUEUE", "exclusive"],
["STRAVA_BACKFILL_QUEUE", "exclusive"],
["GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE", "exclusive"],
["FITBIT_SLEEP_REPAIR_QUEUE", "exclusive"],
["SLEEP_TIMELINE_BACKFILL_QUEUE", "exclusive"],
["LAB_BIOMARKER_BACKFILL_QUEUE", "exclusive"],
])("%s is %s", (queue, policy) => {