From ce41d5bf1729ee29e12e704ec0727607417f6af3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 25 Jul 2026 05:56:42 +0200 Subject: [PATCH] fix(fitbit): key sleep segments on their start instant and sweep re-score leftovers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fitbit re-scores a night after the fact, and the per-segment externalId carried a positional index plus the stage label. Both moved on a re-score: renumbered segments minted fresh keys, the previous scoring's rows stayed live because nothing swept them, and the night total counted both copies from then on. Key the segment on the stable session anchor plus its own start instant (`:sleep:`), and add the replace-by-window sweep every other segment-writing provider already has. The two ship together with the natural-key rescue that has been sitting in the Fitbit upsert unused: the sweep tombstones the old-keyed rows first, so the rescue re-keys those very rows in place instead of inserting a parallel copy or losing the insert to `skipDuplicates` against the natural unique index. A one-shot repair queue re-reads the bounded sleep history once per connection so no user has to do anything; it stamps `sleepRepairedAt` (migration 0270) only on a clean run, never touches the sync watermark, and reports the rows it removed. Nights older than the 365-day backfill horizon keep their old keys and whatever duplicates they already hold — they can no longer gain new ones. Alongside: `fitbitGet` now honours a `Retry-After` inside a 120 s cap by waiting once and retrying once, and fails fast above it so a long hold cannot starve the rest of the poll cohort or outlive a job's expiration. The measurement upsert skips a write whose payload is unchanged, which takes the hourly overlap off the delta feed for every Fitbit resource. --- .../migration.sql | 4 + prisma/schema.prisma | 4 + src/lib/fitbit/client.ts | 158 +++++++++++--- src/lib/fitbit/mapping.md | 14 +- src/lib/fitbit/response-classifier.ts | 14 +- src/lib/fitbit/sync-core.ts | 148 ++++++++++++- src/lib/fitbit/sync-sleep.ts | 36 +++- src/lib/jobs/fitbit-sleep-repair.ts | 201 ++++++++++++++++++ .../jobs/integration-backfill-admission.ts | 3 + .../reminder/register-integration-sync.ts | 72 ++++++- 10 files changed, 606 insertions(+), 48 deletions(-) create mode 100644 prisma/migrations/0270_fitbit_sleep_repair_marker/migration.sql create mode 100644 src/lib/jobs/fitbit-sleep-repair.ts 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/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/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, + ); + } }