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, + ); + } }