diff --git a/CHANGELOG.md b/CHANGELOG.md index 8bfa5c7be..1b50f4df7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,35 @@ ## [Unreleased] +## [1.32.29] — 2026-07-25 + +Fitbit re-scores a night hours after you wake up: blocks get split, merged or +relabelled. HealthLog identified each block by its position in the list, so a +re-score that renumbered the blocks looked like a set of brand new ones. The +previous version stayed on file, nothing removed it, and from then on the night +counted twice. It was quiet: the sync kept reporting success, because the sync +was in fact working. + +- **A re-scored night no longer double counts.** Each sleep block is now + identified by the moment it starts, which does not move when Fitbit rescores + the night. A block that changes from light to deep updates in place instead + of leaving the old copy behind, and anything the new scoring dropped is + cleared from that night's window before the new blocks are written. +- **Nights already affected repair themselves.** A one off pass runs once per + connection after the update and rereads the past year of sleep, so the + duplicates collapse without you doing anything. It covers 365 days, the same + horizon the Fitbit history import uses. Nights older than that keep whatever + they already hold; they cannot gain new duplicates, since Fitbit does not + rescore year old nights. Migration `0270_fitbit_sleep_repair_marker` records + which connections have been through the pass. +- **Rate limits are handled instead of ignored.** When Fitbit asks for a short + wait, the request now waits and tries again once. A long wait is refused + outright and left to the next scheduled run, so one throttled account cannot + hold up everybody else's. +- **Fewer pointless writes.** Fitbit rereads the last 24 hours every hour. Rows + that came back unchanged were rewritten anyway, which showed up as needless + traffic for the iOS app. Unchanged rows are now left alone. + ## [1.32.28] — 2026-07-25 Oura used to look back exactly one week and no further. If the sync had been diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index ddff6a25d..0e84cb9ff 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.32.28 + version: 1.32.29 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index e19716cb8..ece582df5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.32.28", + "version": "1.32.29", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/prisma/migrations/0270_fitbit_sleep_repair_marker/migration.sql b/prisma/migrations/0270_fitbit_sleep_repair_marker/migration.sql new file mode 100644 index 000000000..1a7838eaf --- /dev/null +++ b/prisma/migrations/0270_fitbit_sleep_repair_marker/migration.sql @@ -0,0 +1,4 @@ +-- One-shot Fitbit sleep duplicate repair (stable per-segment sleep keys). +-- Null = repair pending; stamped once the boot-time bounded sleep re-read +-- completes for the connection. +ALTER TABLE "fitbit_connections" ADD COLUMN "sleep_repaired_at" TIMESTAMP(3); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4aadf4f78..b503eb526 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -3655,6 +3655,10 @@ model FitbitConnection { /// Set once the self-converging history backfill walks every collection to /// completion. Null = backfill still in flight (or never started). backfillCompletedAt DateTime? @map("backfill_completed_at") + /// Set once the one-shot sleep repair re-reads the bounded sleep history and + /// collapses the duplicates the earlier volatile segment key left behind. + /// Null = repair pending. + sleepRepairedAt DateTime? @map("sleep_repaired_at") createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") diff --git a/public/sw.js b/public/sw.js index 207db6e90..ca9eabcfc 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.32.28"; + /* @sw-version-fallback */ "v1.32.29"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` 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/fitbit/client.ts b/src/lib/fitbit/client.ts index 040c13563..58f409fdd 100644 --- a/src/lib/fitbit/client.ts +++ b/src/lib/fitbit/client.ts @@ -384,12 +384,51 @@ export function fitbitDate(d: Date): string { return d.toISOString().slice(0, 10); } -/** GET a classic Fitbit endpoint and return the parsed JSON body. */ -async function fitbitGet( +/** + * Longest `Retry-After` delay (seconds) `fitbitGet` will sit out in-process + * before retrying a rate-limited request. + * + * Fitbit's per-user budget resets on the hour, so a 429 raised near the reset + * boundary carries a small `Retry-After` that a bounded wait turns into a + * completed walk. A large value (an exhausted budget, up to a full hour) must + * NOT be slept out: the poll cohort walks users sequentially, so a long hold + * starves every other user in the tick, and inside a pg-boss job the default + * expiration would lap the wait and admit a second execution. Above the cap the + * request fails fast — the classified transient rides the ledger, the watermark + * holds, and the next scheduled tick refetches the window. + */ +export const FITBIT_RETRY_AFTER_WAIT_CAP_S = 120; + +/** + * Read `Retry-After` in its delay-seconds form. Fitbit sends the seconds form; + * the HTTP-date form is deliberately ignored (returns undefined → fail fast) + * rather than parsed into a wait of unknown length. + */ +function parseRetryAfterSeconds(res: Response): number | undefined { + const raw = res.headers.get("retry-after"); + if (typeof raw !== "string") return undefined; + const trimmed = raw.trim(); + if (!/^\d+$/.test(trimmed)) return undefined; + const seconds = Number.parseInt(trimmed, 10); + return Number.isFinite(seconds) ? seconds : undefined; +} + +function waitSeconds(seconds: number): Promise { + return new Promise((resolve) => setTimeout(resolve, seconds * 1000)); +} + +interface FitbitGetAttempt { + json: unknown; + verdict: ReturnType; + retryAfterSeconds: number | undefined; +} + +/** One GET attempt against a classic Fitbit endpoint, classified but not thrown. */ +async function fitbitGetOnce( path: string, accessToken: string, verb: string, -): Promise { +): Promise { const start = performance.now(); const res = await safeFetch(`${FITBIT_API_BASE}${path}`, { method: "GET", @@ -400,22 +439,58 @@ async function fitbitGet( }); const json = await res.json().catch(() => null); const verdict = classifyFitbitResponse(res.status); + const retryAfterSeconds = + res.status === 429 ? parseRetryAfterSeconds(res) : undefined; getEvent()?.addExternalCall({ service: "fitbit", method: verb, duration_ms: Math.round(performance.now() - start), status: res.status, - error: verdict.classification === "success" ? undefined : verdict.reason, + error: + verdict.classification === "success" + ? undefined + : retryAfterSeconds === undefined + ? verdict.reason + : `${verdict.reason} retry_after=${retryAfterSeconds}`, }); - if (verdict.classification !== "success") { - throw new FitbitApiError({ + return { json, verdict, retryAfterSeconds }; +} + +/** + * GET a classic Fitbit endpoint and return the parsed JSON body. + * + * Honours `Retry-After` on a 429 within `FITBIT_RETRY_AFTER_WAIT_CAP_S`: the + * request waits out the delay and retries ONCE. A missing header, a delay above + * the cap, or a second 429 throws exactly as an unhandled failure would — the + * classified error carries `retryAfterSeconds` so the wide event records what + * the upstream asked for. + */ +async function fitbitGet( + path: string, + accessToken: string, + verb: string, +): Promise { + const toError = (attempt: FitbitGetAttempt): FitbitApiError => + new FitbitApiError({ verb, - classification: verdict.classification, - httpStatus: verdict.httpStatus, - reason: verdict.reason, + classification: attempt.verdict.classification, + httpStatus: attempt.verdict.httpStatus, + reason: attempt.verdict.reason, + retryAfterSeconds: attempt.retryAfterSeconds, }); + + const first = await fitbitGetOnce(path, accessToken, verb); + if (first.verdict.classification === "success") return first.json; + + const wait = first.retryAfterSeconds; + if (wait === undefined || wait > FITBIT_RETRY_AFTER_WAIT_CAP_S) { + throw toError(first); } - return json; + + await waitSeconds(wait); + const second = await fitbitGetOnce(path, accessToken, verb); + if (second.verdict.classification === "success") return second.json; + throw toError(second); } // ─── Field → Measurement mapping ─────────────────────────────── @@ -995,27 +1070,57 @@ function parseLocalInstant(iso: string, tz?: string): Date | null { } /** - * Map one Fitbit sleep session into per-SEGMENT `SLEEP_DURATION` rows. Each - * segment's `dateTime` is its START; END = START + seconds. value = minutes. - * Each segment carries a stage-scoped, INDEXED fieldTag so the several segments - * of one stage stay distinct under the `(userId, type, source, externalId)` - * dedup key. Unknown stage labels are skipped. + * One mapped Fitbit sleep session: the stable anchor, the session's covered time + * window (earliest segment start → latest segment end, both UTC), and the + * per-segment rows. The window drives the sync's replace-by-window cleanup of + * stale rows a re-score orphaned; `rows` is what gets upserted. + */ +export interface FitbitSleepSessionMapped { + /** Stable session anchor — also the `:sleep:` externalId prefix. */ + anchor: string; + /** Earliest segment start across the session (UTC), or null if none map. */ + windowStart: Date | null; + /** Latest segment end across the session (UTC), or null if none map. */ + windowEnd: Date | null; + rows: FitbitMappedMeasurement[]; +} + +/** + * Map one Fitbit sleep session into per-SEGMENT `SLEEP_DURATION` rows plus its + * covered window. Each segment's `dateTime` is its START; END = START + seconds. + * value = minutes. Unknown stage labels are skipped; a session with no parseable + * segment yields empty `rows` (and a null window). + * + * Each segment's fieldTag keys off the STABLE session anchor plus the segment's + * own START instant — `:sleep:` — NOT a positional index + * and NOT the stage label. This is the fix for the re-score duplication: the + * index renumbered (and a stage-scoped tag changed) whenever Fitbit re-scored a + * night after the fact, minting fresh externalIds so the upsert created parallel + * rows the night total then double-counted, with no sweep to clear the strays. A + * segment's start instant is stable per block, and dropping the stage from the + * key means a mere re-classification (light→deep on the same block) UPDATES the + * row in place rather than orphaning it. Two segments of one session cannot + * share a start instant, so the key stays unique within the session; across + * sessions the logId anchor separates them. * * The 1.2 sleep log emits OFFSET-LESS local wall-clock timestamps, so `tz` (the * user's stored zone) anchors them to the correct UTC instant — without it a * non-UTC user's near-midnight segment END would shift by their UTC offset and * could land on the wrong wake-day in the night reconstruction. */ -export function mapSleepSession( +export function mapSleepSessionDetailed( session: FitbitSleepSession, tz?: string, -): FitbitMappedMeasurement[] { - const data = session.levels?.data; - if (!Array.isArray(data) || data.length === 0) return []; - +): FitbitSleepSessionMapped { const anchor = sleepSessionAnchor(session, tz); - const out: FitbitMappedMeasurement[] = []; - let segIndex = 0; + const data = session.levels?.data; + const rows: FitbitMappedMeasurement[] = []; + let windowStart: Date | null = null; + let windowEnd: Date | null = null; + if (!Array.isArray(data) || data.length === 0) { + return { anchor, windowStart, windowEnd, rows }; + } + for (const seg of data) { const stage = mapFitbitSleepStage(seg.level); if (!stage) continue; @@ -1026,17 +1131,18 @@ export function mapSleepSession( if (!start) continue; const minutes = seg.seconds / 60; const end = new Date(start.getTime() + seg.seconds * 1000); - out.push({ + if (!windowStart || start < windowStart) windowStart = start; + if (!windowEnd || end > windowEnd) windowEnd = end; + rows.push({ type: "SLEEP_DURATION", value: round2(minutes), unit: "minutes", measuredAt: end, - fieldTag: `${anchor}:sleep_${stage.toLowerCase()}:${segIndex}`, + fieldTag: `${anchor}:sleep:${start.toISOString()}`, sleepStage: stage, }); - segIndex += 1; } - return out; + return { anchor, windowStart, windowEnd, rows }; } /** Read the sleep-session array off the 1.2 sleep-log envelope. */ diff --git a/src/lib/fitbit/mapping.md b/src/lib/fitbit/mapping.md index 5f73a452c..6b554c9c1 100644 --- a/src/lib/fitbit/mapping.md +++ b/src/lib/fitbit/mapping.md @@ -66,10 +66,14 @@ Scope: `sleep`. `/1.2/user/-/sleep/date/{s}/{e}.json` → row **per segment**, `measuredAt = segment START + seconds = segment END`, value = seconds → minutes. The 1.2 log carries a real per-segment series, so the rows lay each block at its true clock time (a MEASURED timeline — rows are NOT flagged -reconstructed). externalId = `:sleep_:` (indexed so several -segments of one stage stay distinct), so a re-scored night overwrites in place. -The segment `dateTime` is local wall-clock ISO without an offset (the night -belongs to the user's local clock) — parsed as local time. +reconstructed). externalId = `:sleep:` — the stable +session anchor plus the segment's own start instant, so a re-scored night +overwrites in place. The key carried a positional index and the stage label +until 2026-07-25; both moved on a re-score, which minted parallel rows the night +total then double-counted, so the key now carries neither and +`replaceStaleFitbitSleep` sweeps the night's window before the fresh set +upserts. The segment `dateTime` is local wall-clock ISO without an offset (the +night belongs to the user's local clock) — parsed as local time. Stage map (stages logs + classic logs): `light → CORE` (Fitbit "light" ↔ Apple "core" shallow-NREM band), `deep → DEEP`, `rem → REM`, `wake`/`awake`/`restless → @@ -112,6 +116,6 @@ ladder). `externalId = :`. Daily-summary rows: `externalId = :`. Daily cumulative activity rows: `externalId = stats::` (Apple-Health overwrite shape). Sleep: -`:sleep_:`. Workouts: `(userId, source: "FITBIT", externalId)` +`:sleep:`. Workouts: `(userId, source: "FITBIT", externalId)` on the `Workout` table. A re-fetch of the same window (daily summaries settle after the fact, so the incremental overlap is 24 h) overwrites in place. diff --git a/src/lib/fitbit/response-classifier.ts b/src/lib/fitbit/response-classifier.ts index 46540168b..765307e92 100644 --- a/src/lib/fitbit/response-classifier.ts +++ b/src/lib/fitbit/response-classifier.ts @@ -53,15 +53,27 @@ export function classifyFitbitResponse( * (message shape `Fitbit error: ` is preserved). */ export class FitbitApiError extends IntegrationApiError { + /** + * The `Retry-After` delay (seconds) the upstream asked for on a 429, when it + * sent the delay-seconds form. Undefined on every other status, on a missing + * header, and on the HTTP-date form. `fitbitGet` waits out a value inside + * `FITBIT_RETRY_AFTER_WAIT_CAP_S` and retries once; anything larger fails + * fast and surfaces here so the wide event records what was asked. + */ + readonly retryAfterSeconds: number | undefined; + constructor(opts: { verb: string; classification: FitbitClassification; httpStatus: number | undefined; reason: string; upstreamError?: string; + retryAfterSeconds?: number; }) { - super({ vendor: "Fitbit", ...opts }); + const { retryAfterSeconds, ...base } = opts; + super({ vendor: "Fitbit", ...base }); this.name = "FitbitApiError"; + this.retryAfterSeconds = retryAfterSeconds; } } diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index d6816ef8e..74637e01b 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/src/lib/fitbit/sync-core.ts @@ -267,6 +267,43 @@ export function noteHardFailure(label: string): void { if (tracker) tracker.failures.push(label); } +/** + * Per-scope tally of rows the sleep replace-by-window sweep tombstoned. The + * count is the deploy-time signal for the one-shot sleep repair (how many + * re-score leftovers a pass actually cleared), and `syncUserSleep` returns only + * its imported count, so the number rides an ambient tracker exactly like the + * soft-skip / hard-fail / rollup-defer ledgers above. + */ +interface SleepSweepTracker { + removed: number; +} +const sleepSweepStorage = new AsyncLocalStorage(); + +/** Record swept sleep rows on the ambient tracker, if one is in scope. */ +export function noteSleepSwept(removed: number): void { + const tracker = sleepSweepStorage.getStore(); + if (tracker) tracker.removed += removed; +} + +/** + * Run `fn` inside a fresh hard-fail ledger + sleep-sweep scope and surface both. + * `syncUserFitbit` owns its own scopes; this export exists for the one-shot jobs + * (sleep repair) that drive a single resource sync directly and must still see a + * dead token or a hard fetch/write failure before stamping their completion + * marker. Ported from Google Health's `runWithGoogleHealthHardFailLedger`, with + * the sweep tally added for the repair's deploy signal. + */ +export async function runWithFitbitHardFailLedger( + fn: () => Promise, +): Promise<{ result: T; failures: string[]; sleepRemoved: number }> { + const tracker: HardFailTracker = { failures: [] }; + const sweep: SleepSweepTracker = { removed: 0 }; + const result = await hardFailStorage.run(tracker, () => + sleepSweepStorage.run(sweep, fn), + ); + return { result, failures: tracker.failures, sleepRemoved: sweep.removed }; +} + /** * Single-source the per-resource collection-fetch error handling. A 403 on one * data class soft-skips it (warn + return 0) so sibling resources still sync; a @@ -399,6 +436,66 @@ export interface FitbitMeasurementUpsert { /** Chunk size for the batched `createMany` insert of fresh readings. */ const FITBIT_CREATE_CHUNK = 500; +/** One just-fetched sleep session's window + the ids that must survive it. */ +export interface FitbitSleepReplaceWindow { + /** Earliest segment start (UTC). Null → nothing to clean for this session. */ + windowStart: Date | null; + /** Latest segment end (UTC). */ + windowEnd: Date | null; + /** The fresh externalIds for THIS session — never soft-deleted. */ + keepIds: string[]; +} + +/** + * Replace-by-window cleanup for re-scored Fitbit sleep sessions. + * + * Fitbit re-scores a night after the fact. Before the stable-key fix a re-fetch + * minted fresh externalIds (the key carried a positional index and the stage + * label) and left the prior scoring's rows LIVE, so the night total silently + * double-counted. For each just-fetched session this soft-deletes any LIVE + * `FITBIT` `SLEEP_DURATION` row whose `measuredAt` falls inside the session's + * `[windowStart, windowEnd]` but was NOT re-produced by this fetch (`keepIds` + * are this session's fresh externalIds). A scan bounded to one session's window + * only ever touches that night's rows — a fresh row is protected by `keepIds`, + * and any leftover (old volatile key, or a segment Fitbit dropped) is cleared. + * Rows OUTSIDE every returned window are never touched, so a night Fitbit did + * not re-report this tick stays intact — no data loss, only stale duplicates and + * re-score orphans go. + * + * ORDERING IS LOAD-BEARING: the caller runs this BEFORE the upsert, so the + * old-keyed rows are tombstoned first and the upsert's natural-key rescue + * re-keys exactly those tombstones in place rather than inserting a parallel + * copy (or losing the insert to `skipDuplicates` against the natural unique + * index). Best-effort — a cleanup failure never fails the user's sync. + */ +export async function replaceStaleFitbitSleep( + userId: string, + sessions: FitbitSleepReplaceWindow[], +): Promise { + let removed = 0; + for (const s of sessions) { + if (!s.windowStart || !s.windowEnd || s.keepIds.length === 0) continue; + try { + const res = await prisma.measurement.updateMany({ + where: { + userId, + source: "FITBIT", + type: "SLEEP_DURATION", + deletedAt: null, + measuredAt: { gte: s.windowStart, lte: s.windowEnd }, + externalId: { notIn: s.keepIds }, + }, + data: { deletedAt: new Date() }, + }); + removed += res.count; + } catch (err) { + getEvent()?.addWarning(`Fitbit: sleep replace-by-window failed: ${err}`); + } + } + noteSleepSwept(removed); + return removed; +} + /** * Write a batch of mapped Fitbit readings for one user and (unless deferred) * fold the rollup tier + invalidate status-insight caches once at the end @@ -453,7 +550,15 @@ export async function upsertFitbitMeasurements( // insert that `skipDuplicates` silently drops against the tombstone's key, // permanently wedging the key (see TOMBSTONES RESURRECT above). const externalIds = readings.map((r) => r.externalId); - let rowByKey = new Map(); + interface ProbedRow { + id: string; + value: number; + unit: string; + measuredAt: Date; + sleepStage: FitbitMeasurementUpsert["sleepStage"]; + deletedAt: Date | null; + } + let rowByKey = new Map(); try { const existing = await prisma.measurement.findMany({ where: { @@ -461,14 +566,34 @@ export async function upsertFitbitMeasurements( source: "FITBIT", externalId: { in: externalIds }, }, - select: { id: true, type: true, externalId: true }, + // The payload fields ride along so the update branch can skip a no-op + // overwrite: the 24 h overlap re-fetches every recent row hourly, and an + // unconditional write would bump `syncVersion` (iOS delta churn) on rows + // whose values did not change. + select: { + id: true, + type: true, + externalId: true, + value: true, + unit: true, + measuredAt: true, + sleepStage: true, + deletedAt: true, + }, }); rowByKey = new Map( existing .filter((e) => e.externalId !== null) .map((e) => [ `${e.type}${DEDUP_KEY_DELIM}${e.externalId}`, - { id: e.id }, + { + id: e.id, + value: e.value, + unit: e.unit, + measuredAt: e.measuredAt, + sleepStage: e.sleepStage, + deletedAt: e.deletedAt, + }, ]), ); } catch (err) { @@ -500,9 +625,20 @@ export async function upsertFitbitMeasurements( for (const r of readings) { const type = r.type as MeasurementType; const key = `${type}${DEDUP_KEY_DELIM}${r.externalId}`; - const live = rowByKey.get(key); - if (live) { - toUpdate.push({ id: live.id, r }); + const existingRow = rowByKey.get(key); + if (existingRow) { + // Skip the no-op overwrite: a LIVE row whose payload matches the + // re-fetched reading exactly gains nothing from a write, and the + // unconditional `syncVersion` bump churned the DB + every iOS delta pull + // hourly for the whole 24 h overlap window. A TOMBSTONED row always + // updates — the write is what resurrects it (`deletedAt: null`). + const unchanged = + existingRow.deletedAt === null && + existingRow.value === r.value && + existingRow.unit === r.unit && + existingRow.measuredAt.getTime() === r.measuredAt.getTime() && + (existingRow.sleepStage ?? null) === (r.sleepStage ?? null); + if (!unchanged) toUpdate.push({ id: existingRow.id, r }); } else if (!plannedCreateKeys.has(key)) { plannedCreateKeys.add(key); toCreate.push({ diff --git a/src/lib/fitbit/sync-sleep.ts b/src/lib/fitbit/sync-sleep.ts index fff450ce6..4bfcb2f45 100644 --- a/src/lib/fitbit/sync-sleep.ts +++ b/src/lib/fitbit/sync-sleep.ts @@ -9,10 +9,12 @@ * real per-segment series, so the rows lay each block at its true clock time (a * MEASURED timeline, not reconstructed). Upserts as `source = FITBIT`. * - * Each segment carries the `sleepStage` axis and an indexed fieldTag so the - * several segments of one stage stay distinct under the dedup key. externalId = - * `:sleep_:` — a re-scored night overwrites in place. A 24 h - * overlap covers Fitbit's after-the-fact re-score. + * Each segment carries the `sleepStage` axis and a fieldTag keyed on the STABLE + * session anchor plus the segment's own start — `:sleep:` + * — so a re-scored night overwrites in place instead of minting parallel rows + * the night total would then double-count. A 24 h overlap covers Fitbit's + * after-the-fact re-score, and `replaceStaleFitbitSleep` clears anything an + * earlier scoring left in the night's window before the fresh set upserts. * * A per-endpoint 403 soft-skips the resource — the `sleep` scope is granted * independently in the consent flow. @@ -20,18 +22,20 @@ import { FITBIT_SLEEP_RANGE_DAYS, fetchSleepRange, - mapSleepSession, + mapSleepSessionDetailed, readSleepSessions, } from "./client"; import { chunkDateRanges, getValidToken, handleCollectionFetchError, + replaceStaleFitbitSleep, upsertFitbitMeasurements, } from "./sync-core"; import type { FitbitMeasurementUpsert, FitbitResourceSyncOptions, + FitbitSleepReplaceWindow, } from "./sync-core"; import { annotate } from "@/lib/logging/context"; import { resolveUserTimezone } from "@/lib/tz/resolver"; @@ -54,6 +58,7 @@ export async function syncUserSleep( const windows = chunkDateRanges(start, end, FITBIT_SLEEP_RANGE_DAYS); const readings: FitbitMeasurementUpsert[] = []; + const replaceWindows: FitbitSleepReplaceWindow[] = []; for (const w of windows) { let body: unknown; try { @@ -62,7 +67,9 @@ export async function syncUserSleep( return handleCollectionFetchError("fetchSleep", userId, err); } for (const session of readSleepSessions(body)) { - for (const m of mapSleepSession(session, tz)) { + const mapped = mapSleepSessionDetailed(session, tz); + if (mapped.rows.length === 0) continue; + for (const m of mapped.rows) { readings.push({ type: m.type, value: m.value, @@ -72,9 +79,22 @@ export async function syncUserSleep( sleepStage: m.sleepStage ?? null, }); } + // Clean any stale rows a prior scoring left in this night's window before + // the fresh set upserts — so a re-scored night reads its true total rather + // than the sum of the old and the re-scored copies. + replaceWindows.push({ + windowStart: mapped.windowStart, + windowEnd: mapped.windowEnd, + keepIds: mapped.rows.map((m) => m.fieldTag), + }); } } + // BEFORE the upsert: the sweep tombstones the old-keyed rows, and the upsert's + // natural-key rescue then re-keys those very tombstones in place. Reversing + // the order would tombstone the rows the upsert just wrote. + const removed = await replaceStaleFitbitSleep(userId, replaceWindows); + const { imported, inserted } = await upsertFitbitMeasurements( userId, readings, @@ -93,6 +113,8 @@ export async function syncUserSleep( .map((r) => r.measuredAt), ).catch(() => {}); - annotate({ action: { name: "fitbit.sleep.sync", details: { imported } } }); + annotate({ + action: { name: "fitbit.sleep.sync", details: { imported, removed } }, + }); return imported; } 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) => { diff --git a/src/lib/jobs/fitbit-sleep-repair.ts b/src/lib/jobs/fitbit-sleep-repair.ts new file mode 100644 index 000000000..13cb7c3f4 --- /dev/null +++ b/src/lib/jobs/fitbit-sleep-repair.ts @@ -0,0 +1,201 @@ +/** + * pg-boss queue + boot-time one-shot sleep duplicate repair for Fitbit + * connections. Modelled on the Google Health sleep repair + * (`src/lib/jobs/google-health-sleep-repair.ts`) and the self-converging + * backfill: a discovery query enqueues one job per connection not yet repaired, + * and the pass is idempotent across reboots (the predicate + * `sleep_repaired_at IS NULL` drops a connection once its repair finishes). + * + * Why: the earlier Fitbit sleep externalId carried a positional segment index + * and the stage label, so every re-score of a night minted fresh keys, stranded + * the previous scoring's rows, and over-counted the night total. The fix + * (`replaceStaleFitbitSleep` plus the stable segment-start key) makes any + * RE-READ night self-heal, and the incremental 24 h overlap heals recent nights + * — but historical nights keep their duplicates until re-read. This job forces + * one bounded sleep-history re-read per connection so the replace-by-window + * collapses each night once. + * + * BOUNDED WALK: unlike the Google repair (whose API pages the whole history), + * the classic Fitbit sleep endpoint caps each request at 30 days and the + * per-user budget is tight, so the repair passes an explicit + * `start = now − FITBIT_BACKFILL_DAYS` — the same 365-day horizon the backfill + * documents as the recovery limit, about a dozen requests per account. Nights + * older than that keep their old keys and any duplicates already recorded; they + * can no longer accrue new ones, because Fitbit does not re-score year-old + * nights. The explicit `start` is REQUIRED: `syncUserSleep`'s own default is + * only 30 days. + * + * Watermark-safe by construction: `syncUserSleep` never reads or stamps + * `lastSyncedAt` — `markSynced` is owned by the orchestrator (`syncUserFitbit`), + * which this job does not call. + * + * The queue name MUST be registered in `allQueues` in + * `src/lib/jobs/reminder/register-integration-sync.ts` or pg-boss never + * provisions it and the boot enqueue silently never drains (the v1.4.37 + * dead-queue class). + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; +import { isReauthRequired } from "@/lib/integrations/status"; +import { + FITBIT_BACKFILL_DAYS, + FITBIT_TOKEN_HARD_FAIL, + runWithFitbitHardFailLedger, +} from "@/lib/fitbit/sync-core"; +import { syncUserSleep } from "@/lib/fitbit/sync-sleep"; + +export const FITBIT_SLEEP_REPAIR_QUEUE = "fitbit-sleep-repair"; + +/** + * Serial concurrency — a repair walks a year of sleep history for one account + * in 30-day requests against the tight per-user budget; concurrency-1 keeps it + * from crowding the request pool, matching `FITBIT_BACKFILL_CONCURRENCY`. + */ +export const FITBIT_SLEEP_REPAIR_CONCURRENCY = 1; + +export interface FitbitSleepRepairPayload { + userId: string; + enqueuedAt: string; +} + +/** + * Per-user repair handler. Re-reads the bounded sleep history for one account — + * `replaceStaleFitbitSleep` collapses each night's leftovers in the write path + * and the natural-key rescue re-keys the survivors — then stamps + * `sleepRepairedAt` so the discovery query drops it. Idempotent: the per-segment + * upserts are key-stable and the replace-by-window clears stale copies, so a + * re-run (e.g. a reboot mid-walk) converges rather than duplicating. Does NOT + * move `lastSyncedAt`. + * + * Verdict-gated: `syncUserSleep` swallows fetch/write hard failures into the + * ambient hard-fail ledger (returning a count either way), so the repair runs it + * inside its own ledger scope — the same mechanism `syncUserFitbit` uses for its + * cycle verdict — and stamps `sleepRepairedAt` ONLY on a clean run. A hard + * fetch/write failure THROWS so pg-boss retries; a dead token (parked + * connection, missing credentials, failed refresh) is NOT clean but does not + * throw — retrying a dead grant burns the retry budget on the same no-op. It + * returns WITHOUT stamping instead: the boot discovery re-enqueues the account + * on the next boot, and the stamp lands after a reconnect. + */ +export async function runFitbitSleepRepairForUser( + userId: string, +): Promise<{ imported: number; removed: number }> { + if (await isReauthRequired(userId, "fitbit")) { + annotate({ + action: { + name: "fitbit.sleepRepair.skipped_reauth", + details: { imported: 0, removed: 0 }, + }, + }); + return { imported: 0, removed: 0 }; + } + + const start = new Date( + Date.now() - FITBIT_BACKFILL_DAYS * 24 * 60 * 60 * 1000, + ); + + const { + result: imported, + failures, + sleepRemoved: removed, + } = await runWithFitbitHardFailLedger(() => + syncUserSleep(userId, { start, deferRollup: false }), + ); + + if (failures.length > 0) { + if (failures.every((f) => f === FITBIT_TOKEN_HARD_FAIL)) { + // Dead token: nothing was fetched, so nothing to retry until the user + // reconnects. No stamp — the next boot's discovery picks it back up. + annotate({ + action: { + name: "fitbit.sleepRepair.skipped_token", + details: { imported, removed }, + }, + }); + return { imported, removed }; + } + throw new Error( + `fitbit sleep repair incomplete for user ${userId} (${failures.join(", ")}) — marker not stamped`, + ); + } + + await prisma.fitbitConnection.update({ + where: { userId }, + data: { sleepRepairedAt: new Date() }, + }); + + annotate({ + action: { + name: "fitbit.sleepRepair.complete", + details: { imported, removed }, + }, + }); + return { imported, removed }; +} + +/** + * Boot-time discovery. Finds every Fitbit connection not yet repaired + * (`sleep_repaired_at IS NULL`) and enqueues one repair job per account. + * + * Idempotent across reboots: once a connection's repair completes, + * `sleepRepairedAt` is set and the predicate drops it from the discovery set. + * pg-boss `singletonKey` coalesces duplicate sends so a fast restart while a job + * is queued doesn't double up. + * + * Best-effort: errors are returned through the result value so the worker boot + * never fails because of a repair miss. + */ +export async function enqueueBootTimeFitbitSleepRepair( + startAfterSeconds: number = 0, +): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + const connections = await prisma.fitbitConnection.findMany({ + where: { sleepRepairedAt: null }, + select: { userId: true }, + }); + + if (connections.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { userId } of connections) { + const payload: FitbitSleepRepairPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send( + FITBIT_SLEEP_REPAIR_QUEUE, + payload, + integrationBackfillSourceOptions( + `fitbit-sleep-repair|${userId}`, + startAfterSeconds, + ), + ); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} diff --git a/src/lib/jobs/integration-backfill-admission.ts b/src/lib/jobs/integration-backfill-admission.ts index 85f5e992e..e00f2d06c 100644 --- a/src/lib/jobs/integration-backfill-admission.ts +++ b/src/lib/jobs/integration-backfill-admission.ts @@ -16,6 +16,9 @@ export const INTEGRATION_BACKFILL_BOOT_ORDER = [ "sleep-timeline-backfill", "lab-biomarker-backfill", "strava-backfill", + // Appended, not slotted next to `fitbit-backfill`, so every existing + // provider's boot stagger stays exactly where it was. + "fitbit-sleep-repair", ] as const; export type IntegrationBackfillKind = diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index 7abc44a0e..cb5e85d74 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -3,9 +3,9 @@ * * Owns the third-party-provider sync queues — Withings (fallback / activity / * sleep), WHOOP (recovery / sleep / workout / cycle + backfill + OAuth-state - * cleanup), Fitbit (sync + backfill + OAuth-state cleanup), Nightscout, Polar, - * Oura, MoodLog, and the two one-shot backfills (sleep-timeline, lab-biomarker) - * — plus their boot-time discovery enqueues. + * cleanup), Fitbit (sync + backfill + sleep repair + OAuth-state cleanup), + * Nightscout, Polar, Oura, MoodLog, and the two one-shot backfills + * (sleep-timeline, lab-biomarker) — plus their boot-time discovery enqueues. * * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`, @@ -42,6 +42,13 @@ import { enqueueBootTimeGoogleHealthSleepRepair, type GoogleHealthSleepRepairPayload, } from "@/lib/jobs/google-health-sleep-repair"; +import { + FITBIT_SLEEP_REPAIR_QUEUE, + FITBIT_SLEEP_REPAIR_CONCURRENCY, + runFitbitSleepRepairForUser, + enqueueBootTimeFitbitSleepRepair, + type FitbitSleepRepairPayload, +} from "@/lib/jobs/fitbit-sleep-repair"; import { STRAVA_BACKFILL_QUEUE, STRAVA_BACKFILL_CONCURRENCY, @@ -285,6 +292,10 @@ const allQueues = [ // self-converging boot backfill, and the daily OAuth-state ledger sweep. FITBIT_SYNC_QUEUE, FITBIT_BACKFILL_QUEUE, + // One-shot Fitbit sleep duplicate repair. Discovery enqueues one bounded + // sleep re-read per connection not yet repaired; the replace-by-window write + // path collapses each night's leftovers. Idempotent across reboots. + FITBIT_SLEEP_REPAIR_QUEUE, FITBIT_OAUTH_STATE_CLEANUP_QUEUE, // v1.26.0 — Google Health poll-only sync (no webhook at launch), // self-converging boot backfill, and the daily OAuth-state ledger sweep. @@ -411,6 +422,11 @@ const queuePolicies: QueuePolicyTable = { reason: "Per-connection one-shot sleep re-read; discovery drops the connection once it is marked repaired.", }, + [FITBIT_SLEEP_REPAIR_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection one-shot sleep re-read; discovery drops the connection once it is marked repaired.", + }, [SLEEP_TIMELINE_BACKFILL_QUEUE]: { policy: "exclusive", reason: @@ -484,6 +500,15 @@ export async function registerIntegrationSyncQueues( ); break; } + case "fitbit-sleep-repair": { + const { imported, removed } = + await runFitbitSleepRepairForUser(userId); + workerLog( + "info", + `[fitbit-sleep-repair] user=${userId} imported=${imported} removed=${removed}`, + ); + break; + } case "sleep-timeline-backfill": { if (!data.provider) { throw new Error( @@ -616,6 +641,21 @@ export async function registerIntegrationSyncQueues( } }, ); + // One-shot Fitbit sleep duplicate repair. This source queue retains the + // per-connection singleton and forwards the watermark-safe bounded sleep + // re-read to the shared drain. + await boss.work( + FITBIT_SLEEP_REPAIR_QUEUE, + { localConcurrency: FITBIT_SLEEP_REPAIR_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + await enqueueIntegrationBackfillAdmission(boss, { + kind: "fitbit-sleep-repair", + data: job.data, + }); + } + }, + ); await boss.work( FITBIT_OAUTH_STATE_CLEANUP_QUEUE, { localConcurrency: 1 }, @@ -923,4 +963,30 @@ export async function enqueueIntegrationSyncBootDiscovery(): Promise { err, ); } + + // One-shot Fitbit sleep duplicate repair. Finds every connection not yet + // repaired and enqueues one bounded sleep re-read per account; a completed + // pass stamps `sleepRepairedAt`. + try { + const { enqueued, skipped, error } = await enqueueBootTimeFitbitSleepRepair( + bootStaggerSecondsFor("fitbit-sleep-repair"), + ); + if (error) { + workerLog( + "error", + `[fitbit-sleep-repair] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[fitbit-sleep-repair] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[fitbit-sleep-repair] boot discovery threw an unexpected error", + err, + ); + } }