From a08c759763451b0dccf08beaeaab3b574a0b77f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 25 Jul 2026 05:56:52 +0200 Subject: [PATCH] test(fitbit): cover the stable sleep key, the sweep, and the one-shot repair MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/lib/fitbit/__tests__/client.test.ts | 246 ++++++++++++- .../__tests__/sync-sleep-replace.test.ts | 341 ++++++++++++++++++ .../__tests__/sync-upsert-resurrect.test.ts | 126 +++++++ src/lib/jobs/__tests__/fitbit-queues.test.ts | 11 + .../__tests__/fitbit-sleep-repair.test.ts | 233 ++++++++++++ .../integration-backfill-admission.test.ts | 6 +- .../__tests__/queue-policy-tables.test.ts | 1 + 7 files changed, 953 insertions(+), 11 deletions(-) create mode 100644 src/lib/fitbit/__tests__/sync-sleep-replace.test.ts create mode 100644 src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts diff --git a/src/lib/fitbit/__tests__/client.test.ts b/src/lib/fitbit/__tests__/client.test.ts index 381073217..c66cb4696 100644 --- a/src/lib/fitbit/__tests__/client.test.ts +++ b/src/lib/fitbit/__tests__/client.test.ts @@ -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; + }>, +) { 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); }); }); diff --git a/src/lib/fitbit/__tests__/sync-sleep-replace.test.ts b/src/lib/fitbit/__tests__/sync-sleep-replace.test.ts new file mode 100644 index 000000000..4586a24c2 --- /dev/null +++ b/src/lib/fitbit/__tests__/sync-sleep-replace.test.ts @@ -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> + >(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("../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; + data: Record; + }; + 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; + }; + 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; + }, + ); + 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); + }); +}); diff --git a/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts b/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts index 69dca3e57..508bf739c 100644 --- a/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts +++ b/src/lib/fitbit/__tests__/sync-upsert-resurrect.test.ts @@ -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; + }; + 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; + }; + expect(upd.data.deletedAt).toBeNull(); + expect(imported).toBe(1); + }); +}); + describe("upsertFitbitMeasurements — natural-key migration rescue", () => { const SLEEP_READING = { type: "SLEEP_DURATION", diff --git a/src/lib/jobs/__tests__/fitbit-queues.test.ts b/src/lib/jobs/__tests__/fitbit-queues.test.ts index f4dd416b8..3e6145540 100644 --- a/src/lib/jobs/__tests__/fitbit-queues.test.ts +++ b/src/lib/jobs/__tests__/fitbit-queues.test.ts @@ -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["']/); diff --git a/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts b/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts new file mode 100644 index 000000000..f35929a20 --- /dev/null +++ b/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts @@ -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; + data: Record; + }; + 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 }; + 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(); + }); +}); diff --git a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts index 461ac2057..ae009c783 100644 --- a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts +++ b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts @@ -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( diff --git a/src/lib/jobs/__tests__/queue-policy-tables.test.ts b/src/lib/jobs/__tests__/queue-policy-tables.test.ts index 213362068..d40ce3af1 100644 --- a/src/lib/jobs/__tests__/queue-policy-tables.test.ts +++ b/src/lib/jobs/__tests__/queue-policy-tables.test.ts @@ -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) => {