diff --git a/prisma/migrations/0424_oura_backfill_progress/migration.sql b/prisma/migrations/0424_oura_backfill_progress/migration.sql new file mode 100644 index 000000000..388a4d940 --- /dev/null +++ b/prisma/migrations/0424_oura_backfill_progress/migration.sql @@ -0,0 +1,13 @@ +-- Where the Oura history backfill has got, and when it finished. +-- +-- The backfill reads every Oura collection's history in slices through the +-- shared integration backfill lane. After each window it writes, the progress +-- note records where the collection's next window ends, so a slice that stops +-- on its budget or a rate limit resumes there instead of from the newest day. +-- The completion stamp drops the account from boot discovery. +-- +-- Both nullable and unset for every existing row: every current Oura +-- connection is offered the backfill at the next boot, with no reconnect. + +ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "oura_backfill_progress" JSONB; +ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "oura_backfill_completed_at" TIMESTAMP(3); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 1d79d97b1..4aa42a5d4 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -277,22 +277,32 @@ model User { // v1.17.0 — Oura OAuth credentials (F4, per-user, encrypted at app level). // Mirrors the WHOOP credential columns; OAuth access/refresh tokens stored // encrypted. - ouraAccessTokenEncrypted String? @map("oura_access_token_encrypted") - ouraRefreshTokenEncrypted String? @map("oura_refresh_token_encrypted") + ouraAccessTokenEncrypted String? @map("oura_access_token_encrypted") + ouraRefreshTokenEncrypted String? @map("oura_refresh_token_encrypted") // v1.17.1 — Oura BYO-app client credentials (W1c, per-user, encrypted at app // level). Mirrors the WHOOP/Fitbit `clientId`/`clientSecret` columns 1:1 so a // self-hoster pastes their own Oura Cloud app's client id/secret into Settings // before the OAuth connect flow runs. Stored encrypted as a string via // `src/lib/crypto.ts`, matching every other `*Encrypted` column. - ouraClientIdEncrypted String? @map("oura_client_id_encrypted") - ouraClientSecretEncrypted String? @map("oura_client_secret_encrypted") + ouraClientIdEncrypted String? @map("oura_client_id_encrypted") + ouraClientSecretEncrypted String? @map("oura_client_secret_encrypted") /// v1.42.3 — Oura collections the current grant does not cover, as a JSON /// string array of collection names (`spo2`, `vo2max`, `cardio_age`, /// `resilience`). Written by the sync when a collection answers 401/403 while /// its siblings succeed with the same token, which is a scope the grant never /// carried, not a dead token. The Oura card reads it to ask for one /// reconnect. Cleared on connect, disconnect and credential removal. - ouraScopeDenied Json? @map("oura_scope_denied") + ouraScopeDenied Json? @map("oura_scope_denied") + /// v1.44 — how far the Oura history backfill has got: per collection, the + /// day its next window ends on, and whether it is done or refused for a + /// scope; plus where the account's history starts once learned. Written + /// after every window so a slice resumes where the last one stopped (see + /// `src/lib/oura/backfill-progress.ts`); kept after completion so a + /// reconnect reads only what the old grant refused. + ouraBackfillProgress Json? @map("oura_backfill_progress") + /// v1.44 — the Oura history backfill finished (self-converging discovery + /// predicate `oura_backfill_completed_at IS NULL`). Cleared on connect. + ouraBackfillCompletedAt DateTime? @map("oura_backfill_completed_at") // v1.28.x — Strava OAuth credentials (per-user BYO-app, encrypted at app // level). Strava caps every newly-created API app at single-player mode diff --git a/src/__tests__/backfill-preserves-updated-at-guard.test.ts b/src/__tests__/backfill-preserves-updated-at-guard.test.ts index 5f3238eb7..39fe550ab 100644 --- a/src/__tests__/backfill-preserves-updated-at-guard.test.ts +++ b/src/__tests__/backfill-preserves-updated-at-guard.test.ts @@ -106,6 +106,8 @@ const BACKFILL_STAMPS: Record = { "marks the connection's one-shot as done; not a listed record", "src/lib/jobs/strava-backfill.ts > user": "stamps the account's backfill-completed marker", + "src/lib/jobs/oura-backfill.ts > user": + "writes the account's backfill progress note and completed marker", "src/lib/jobs/document-content-index-backfill.ts > documentContentIndex": "the index row is rebuilt content; its updatedAt is the rebuild time", "src/lib/jobs/lab-biomarker-backfill.ts > biomarker": diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 3a72f236a..06f012349 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -53,6 +53,8 @@ const WRITE_SITES: Record = { "lib/whoop/webhook-handler.ts": TAIL, // Retiring the rows of Oura sleep periods the person rejected or deleted. "lib/oura/sync.ts": TAIL, + // Retiring the equivalent-distance rows earlier Oura syncs wrote. + "lib/oura/retire-distance.ts": TAIL, // Integration syncs: each collects the (type, day) pairs it touched and // runs both legs itself at the end of the sync. "lib/fitbit/sync-core.ts": diff --git a/src/app/api/oura/callback/__tests__/route.test.ts b/src/app/api/oura/callback/__tests__/route.test.ts index 2806299e4..2763e6577 100644 --- a/src/app/api/oura/callback/__tests__/route.test.ts +++ b/src/app/api/oura/callback/__tests__/route.test.ts @@ -206,12 +206,17 @@ describe("GET /api/oura/callback — issuer", () => { * card for up to an hour. */ describe("GET /api/oura/callback — connect-time first sync", () => { - it("enqueues exactly one targeted oura-sync job for the connecting user", async () => { + it("enqueues one targeted oura-sync job and the history backfill for the connecting user", async () => { const res = await GET(makeReq({ code: "c", state: "S", cookie: "S" })); expect(res.headers.get("location")).toContain("oura=connected"); - expect(bossSend).toHaveBeenCalledTimes(1); + expect(bossSend).toHaveBeenCalledTimes(2); expect(bossSend).toHaveBeenCalledWith("oura-sync", { userId: "u1" }); + expect(bossSend).toHaveBeenCalledWith( + "oura-backfill", + expect.objectContaining({ userId: "u1" }), + expect.objectContaining({ singletonKey: "oura-backfill|u1" }), + ); }); it("still connects when no boss instance is available", async () => { diff --git a/src/app/api/oura/callback/route.ts b/src/app/api/oura/callback/route.ts index 6e7b7ed20..7edc6172b 100644 --- a/src/app/api/oura/callback/route.ts +++ b/src/app/api/oura/callback/route.ts @@ -17,6 +17,11 @@ import { ouraCollectionsMissingFromGrant } from "@/lib/oura/scope"; import { markReconnected } from "@/lib/integrations/status"; import { OURA_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; +import { + OURA_BACKFILL_QUEUE, + type OuraBackfillPayload, +} from "@/lib/jobs/oura-backfill"; import { NextRequest, NextResponse } from "next/server"; /** @@ -134,6 +139,10 @@ async function completeCallback( ouraRefreshTokenEncrypted: encrypt(tokens.refresh_token), ouraScopeDenied: missing && missing.length > 0 ? missing : Prisma.DbNull, + // Open the history backfill again: a first connect has none yet, + // and a reconnect may grant what the old grant refused. What was + // already walked stays in the progress note and is not read again. + ouraBackfillCompletedAt: null, }, }); @@ -153,6 +162,21 @@ async function completeCallback( .catch((err) => getEvent()?.addWarning(`oura-sync connect enqueue failed: ${err}`), ); + const backfill: OuraBackfillPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + await boss + .send( + OURA_BACKFILL_QUEUE, + backfill, + integrationBackfillSourceOptions(`oura-backfill|${userId}`), + ) + .catch((err) => + getEvent()?.addWarning( + `oura-backfill connect enqueue failed: ${err}`, + ), + ); } const response = NextResponse.redirect( diff --git a/src/lib/data-wipe/wipe-plan.ts b/src/lib/data-wipe/wipe-plan.ts index f02b0d0e7..e38c3dc79 100644 --- a/src/lib/data-wipe/wipe-plan.ts +++ b/src/lib/data-wipe/wipe-plan.ts @@ -463,6 +463,8 @@ export const USER_RESET = { ouraClientIdEncrypted: null, ouraClientSecretEncrypted: null, ouraScopeDenied: Prisma.DbNull, + ouraBackfillProgress: Prisma.DbNull, + ouraBackfillCompletedAt: null, stravaClientIdEncrypted: null, stravaClientSecretEncrypted: null, stravaAccessTokenEncrypted: null, diff --git a/src/lib/export/backup-plan.ts b/src/lib/export/backup-plan.ts index b3c9ddef3..8b17b24ee 100644 --- a/src/lib/export/backup-plan.ts +++ b/src/lib/export/backup-plan.ts @@ -959,6 +959,10 @@ export const USER_COLUMN_BACKUP_CLASS = { stravaLastActivityAt: "OPERATIONAL", stravaBackfillCompletedAt: "OPERATIONAL", stravaBackfillResumeBefore: "OPERATIONAL", + // The restored account's Oura backfill reads its history again from the + // provider; a carried note would claim history this host never imported. + ouraBackfillProgress: "OPERATIONAL", + ouraBackfillCompletedAt: "OPERATIONAL", healthKitLastSyncedAt: "OPERATIONAL", healthKitLastSyncTrigger: "OPERATIONAL", healthKitLastBackgroundSyncAt: "OPERATIONAL", diff --git a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts index 71eededb5..a26bd6561 100644 --- a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts +++ b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts @@ -28,6 +28,7 @@ const providerSources = [ "lab-biomarker-backfill.ts", "strava-backfill.ts", "fitbit-sleep-repair.ts", + "oura-backfill.ts", ].map((file) => readFileSync(join(__dirname, "..", file), "utf8")); describe("shared full-history backfill admission", () => { @@ -41,9 +42,10 @@ describe("shared full-history backfill admission", () => { "lab-biomarker-backfill", "strava-backfill", "fitbit-sleep-repair", + "oura-backfill", ]); expect(INTEGRATION_BACKFILL_BOOT_ORDER.map(bootStaggerSecondsFor)).toEqual([ - 30, 60, 90, 120, 150, 180, 210, 240, + 30, 60, 90, 120, 150, 180, 210, 240, 270, ]); expect(INTEGRATION_BACKFILL_STAGGER_SECONDS).toBe(30); @@ -61,6 +63,7 @@ describe("shared full-history backfill admission", () => { "lab-biomarker-backfill": "enqueueBootTimeLabBiomarkerBackfill", "strava-backfill": "enqueueBootTimeStravaBackfill", "fitbit-sleep-repair": "enqueueBootTimeFitbitSleepRepair", + "oura-backfill": "enqueueBootTimeOuraBackfill", } as const; for (const kind of INTEGRATION_BACKFILL_BOOT_ORDER) { expect(registrarSource).toMatch( @@ -153,6 +156,7 @@ describe("shared full-history backfill admission", () => { "LAB_BIOMARKER_BACKFILL_QUEUE", "STRAVA_BACKFILL_QUEUE", "FITBIT_SLEEP_REPAIR_QUEUE", + "OURA_BACKFILL_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 128c73dad..bc4b5ea04 100644 --- a/src/lib/jobs/__tests__/queue-policy-tables.test.ts +++ b/src/lib/jobs/__tests__/queue-policy-tables.test.ts @@ -90,6 +90,7 @@ describe("integration-sync registrar queue policies", () => { ["FITBIT_BACKFILL_QUEUE", "exclusive"], ["GOOGLE_HEALTH_BACKFILL_QUEUE", "exclusive"], ["STRAVA_BACKFILL_QUEUE", "exclusive"], + ["OURA_BACKFILL_QUEUE", "exclusive"], ["GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE", "exclusive"], ["FITBIT_SLEEP_REPAIR_QUEUE", "exclusive"], ["SLEEP_TIMELINE_BACKFILL_QUEUE", "exclusive"], diff --git a/src/lib/jobs/integration-backfill-admission.ts b/src/lib/jobs/integration-backfill-admission.ts index 382a65374..a2f1b05a6 100644 --- a/src/lib/jobs/integration-backfill-admission.ts +++ b/src/lib/jobs/integration-backfill-admission.ts @@ -19,6 +19,7 @@ export const INTEGRATION_BACKFILL_BOOT_ORDER = [ // Appended, not slotted next to `fitbit-backfill`, so every existing // provider's boot stagger stays exactly where it was. "fitbit-sleep-repair", + "oura-backfill", ] as const; export type IntegrationBackfillKind = diff --git a/src/lib/jobs/integration-backfill-drain.ts b/src/lib/jobs/integration-backfill-drain.ts index 15ed48fdf..dbce7a548 100644 --- a/src/lib/jobs/integration-backfill-drain.ts +++ b/src/lib/jobs/integration-backfill-drain.ts @@ -22,6 +22,7 @@ import { runGoogleHealthBackfillForUser } from "@/lib/jobs/google-health-backfil import { runGoogleHealthSleepRepairForUser } from "@/lib/jobs/google-health-sleep-repair"; import { runFitbitSleepRepairForUser } from "@/lib/jobs/fitbit-sleep-repair"; import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill"; +import { runOuraBackfillForUser } from "@/lib/jobs/oura-backfill"; import { runSleepTimelineBackfillForUser } from "@/lib/jobs/sleep-timeline-backfill"; import { runLabBiomarkerBackfillForUser } from "@/lib/jobs/lab-biomarker-backfill"; import { @@ -144,6 +145,20 @@ async function runAdmitted( ); return; } + case "oura-backfill": { + const { imported, continued } = await runOuraBackfillForUser( + userId, + shouldStop, + data.slice ?? 0, + ); + totals.imported += imported; + if (continued) totals.continued += 1; + workerLog( + "info", + `[oura-backfill] user=${userId} imported=${imported}${continued ? " continued" : ""}`, + ); + return; + } } } diff --git a/src/lib/jobs/oura-backfill.ts b/src/lib/jobs/oura-backfill.ts new file mode 100644 index 000000000..6fad28bef --- /dev/null +++ b/src/lib/jobs/oura-backfill.ts @@ -0,0 +1,516 @@ +/** + * v1.44 — the Oura history backfill: every collection's history, read once, + * in resumable slices through the shared integration backfill lane. + * + * The hourly sync reads a window of days (hours for heart rate); everything + * older is this job's. Discovery finds every Oura connection whose + * `ouraBackfillCompletedAt` is unset (an existing connection included: the + * column arrives empty, so no reconnect is needed), and the slices walk each + * collection from the day the backfill began back to the start of the + * account's history, window by window, noting after every window where they + * got to (`../oura/backfill-progress.ts`). A slice ends on its own budget, on + * the job's, when another heavy job wants the lane, or on Oura's rate limit, + * and the next slice picks up at the window it stopped before. A collection + * is never read again from the anchor once it has moved. + * + * Memory: a window of a day-keyed collection is at most ninety days of + * records; heart rate is streamed a page at a time (`syncOuraHeartRateRange`) + * and keeps only the days it touched. Nothing accumulates across windows. + * + * The first thing a backfill does, once per account, is retire the + * equivalent-distance rows earlier syncs wrote (`retireOuraEquivalentDistance`). + * An account that holds such rows but no longer has an Oura connection is + * offered too: it retires them and is stamped complete, and a later connect + * reopens the backfill (the callback clears the stamp). + * + * The marker is stamped only when every collection is done or refused for a + * scope the grant lacks; the progress note stays beside it, so a reconnect + * that grants the missing scope reads only those collections: the first slice + * of every chain asks again for the refused ones. + */ +import { prisma } from "@/lib/db"; +import { annotate, getEvent } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + enqueueIntegrationBackfillAdmission, + INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + integrationBackfillSourceOptions, +} from "@/lib/jobs/integration-backfill-admission"; +import { LANE_YIELD_DELAY_SECONDS, laneWanted } from "@/lib/jobs/heavy-lane"; +import { reportJobProgress } from "@/lib/jobs/job-observer"; +import { isReauthRequired } from "@/lib/integrations/status"; +import { shiftDateKey, userDayKey } from "@/lib/tz/format"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; +import { startOfLocalDayKey } from "@/lib/tz/local-day"; +import { sweepStaleSleepSegments } from "@/lib/sleep/sweep-stale-segments"; +import { getOuraConnection } from "@/lib/oura/credentials"; +import { refreshOuraAfterRefusal } from "@/lib/oura/oauth-descriptor"; +import { OuraApiError } from "@/lib/oura/response-classifier"; +import { OURA_COLLECTIONS, type OuraCollectionName } from "@/lib/oura/scope"; +import { + isNamedScopeRefusal, + isTokenRefusal, + ouraCollectionReader, + ouraRunContext, + recordOuraSyncFailure, + retireOuraSleepRecords, + upsertOuraMeasurements, +} from "@/lib/oura/sync"; +import { refoldOuraTouchedDays } from "@/lib/oura/sync-heart-rate"; +import { + OURA_EQUIVALENT_DISTANCE_WHERE, + retireOuraEquivalentDistance, +} from "@/lib/oura/retire-distance"; +import { + OURA_HISTORY_FLOOR, + ouraBackfillProgressJson, + readOuraBackfillProgress, + withoutDenied, + type OuraBackfillProgress, +} from "@/lib/oura/backfill-progress"; + +export const OURA_BACKFILL_QUEUE = "oura-backfill"; + +/** One account at a time; the shared lane bounds the instance. */ +export const OURA_BACKFILL_CONCURRENCY = 1; + +/** + * One slice: five minutes of walking, then the account is let go and the + * rest is the next slice's. Short so a slice never sits long on the lane. + */ +export const OURA_BACKFILL_SLICE_MS = 5 * 60 * 1000; + +/** The pause before the next slice after a rate limit that named no wait. */ +export const OURA_BACKFILL_THROTTLED_DELAY_SECONDS = 5 * 60; + +/** + * The highest slice a chain may reach. Eleven years of history is about 600 + * heart-rate windows and a few hundred more for the rest, and a slice writes + * dozens of windows; two thousand slices bounds nothing a working backfill + * reaches, and stops one that counts progress it does not make. + */ +export const OURA_BACKFILL_MAX_SLICE = 2000; + +/** Days per window, by collection: dense or heavy ones read shorter windows. */ +const WINDOW_DAYS: Partial> = { + sleep: 30, + heartrate: 7, +}; +const DEFAULT_WINDOW_DAYS = 90; + +/** The collections that find where history starts; walked first. */ +const HISTORY_PROBES: readonly OuraCollectionName[] = ["activity", "sleep"]; + +/** How far before the learned history start the other collections reach. */ +const HISTORY_MARGIN_DAYS = 7; + +/** Walk order: the history probes first, heart rate (the dense one) last. */ +export const OURA_BACKFILL_ORDER: readonly OuraCollectionName[] = [ + ...HISTORY_PROBES, + ...OURA_COLLECTIONS.map((c) => c.name).filter( + (n) => !HISTORY_PROBES.includes(n) && n !== "heartrate", + ), + "heartrate", +]; + +export interface OuraBackfillPayload { + userId: string; + enqueuedAt: string; +} + +/** Test seam: a slice's budget and clock. */ +export interface OuraBackfillSliceOptions { + sliceMs?: number; + now?: () => number; +} + +type StopReason = "budget" | "lane_wanted" | "rate_limited"; + +function minDay(a: string | undefined, b: string): string { + return a === undefined || b < a ? b : a; +} + +/** The floor a collection walks back to, or undefined while not yet known. */ +function floorFor( + name: OuraCollectionName, + progress: OuraBackfillProgress, +): string | undefined { + if (HISTORY_PROBES.includes(name)) return OURA_HISTORY_FLOOR; + if (progress.historyStart === undefined) return undefined; + // No activity and no sleep anywhere: there is no history to read. + if (progress.historyStart === null) return progress.anchor; + const margin = shiftDateKey(progress.historyStart, -HISTORY_MARGIN_DAYS); + return margin < OURA_HISTORY_FLOOR ? OURA_HISTORY_FLOOR : margin; +} + +/** Settle `historyStart` once both probes are finished. */ +function settleHistoryStart(progress: OuraBackfillProgress): void { + if (progress.historyStart !== undefined) return; + const finished = HISTORY_PROBES.every((n) => { + const s = progress.collections[n]; + return s?.done || s?.denied; + }); + if (finished) progress.historyStart = progress.oldestSeen ?? null; +} + +async function saveProgress( + userId: string, + progress: OuraBackfillProgress, +): Promise { + await prisma.user.update({ + where: { id: userId }, + data: { ouraBackfillProgress: ouraBackfillProgressJson(progress) }, + }); +} + +/** + * Read and write one window of one collection. Returns the oldest day a + * history probe returned, if any. + */ +async function runWindow( + userId: string, + tz: string, + name: OuraCollectionName, + window: { startDate: string; endDate: string }, + token: string, +): Promise<{ oldestDay?: string; imported: number }> { + const ctx = ouraRunContext( + userId, + tz, + window, + // The heart-rate window covers the same local days, end day included. + { + start: startOfLocalDayKey(window.startDate, tz), + end: startOfLocalDayKey(shiftDateKey(window.endDate, 1), tz), + }, + false, + ); + const readings = await ouraCollectionReader(name, ctx)(token); + await sweepStaleSleepSegments(userId, "OURA", ctx.sleepSweeps); + await retireOuraSleepRecords(userId, ctx.sleepRetirements); + const written = await upsertOuraMeasurements(userId, readings, { + emitArrivals: false, + }); + await refoldOuraTouchedDays(userId, ctx.heartRateTouched); + let oldestDay: string | undefined; + if (HISTORY_PROBES.includes(name)) { + for (const r of readings) { + oldestDay = minDay(oldestDay, userDayKey(r.measuredAt, tz)); + } + // A probe that only retired periods still saw history. + if (ctx.sleepRetirements.length > 0 || ctx.sleepSweeps.length > 0) { + oldestDay = minDay(oldestDay, window.startDate); + } + } + return { oldestDay, imported: written + ctx.sideImported }; +} + +/** + * Run one slice of the backfill for one account. Returns how many rows it + * wrote and whether it sent the next slice. + */ +export async function runOuraBackfillForUser( + userId: string, + shouldStop: () => boolean = () => false, + slice: number = 0, + options: OuraBackfillSliceOptions = {}, +): Promise<{ imported: number; continued: boolean }> { + if (slice > OURA_BACKFILL_MAX_SLICE) { + throw new Error( + `oura backfill for user ${userId} reached slice ${slice}, past the ceiling of ${OURA_BACKFILL_MAX_SLICE}`, + ); + } + const now = options.now ?? Date.now; + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { ouraBackfillProgress: true, ouraBackfillCompletedAt: true }, + }); + if (!user || user.ouraBackfillCompletedAt) { + annotate({ + action: { name: "oura.backfill.already_complete", details: { slice } }, + }); + return { imported: 0, continued: false }; + } + + const tz = await resolveUserTimezone(userId); + const stored = readOuraBackfillProgress( + user.ouraBackfillProgress, + OURA_BACKFILL_ORDER, + ); + // The first slice of a chain (a boot's, or a connect's) asks again for + // what an earlier grant refused: the grant may cover it now. One request + // each finds out; everything already walked stays walked. + const progress: OuraBackfillProgress = stored + ? slice === 0 + ? withoutDenied(stored) + : stored + : { anchor: userDayKey(new Date(now()), tz), collections: {} }; + + if (!progress.distanceRetired) { + const retired = await retireOuraEquivalentDistance(userId); + progress.distanceRetired = true; + await saveProgress(userId, progress); + annotate({ meta: { "oura.backfill.distance_retired": retired } }); + } + + const conn = await getOuraConnection(userId); + if (!conn) { + // Nothing to read: the account only had legacy rows to retire. A later + // connect reopens the backfill. + await prisma.user.update({ + where: { id: userId }, + data: { ouraBackfillCompletedAt: new Date(now()) }, + }); + return { imported: 0, continued: false }; + } + if (await isReauthRequired(userId, "oura")) { + annotate({ + action: { name: "oura.backfill.skipped_reauth", details: { slice } }, + }); + return { imported: 0, continued: false }; + } + + const deadline = now() + (options.sliceMs ?? OURA_BACKFILL_SLICE_MS); + const knownDenied = new Set(conn.scopeDenied ?? []); + let token = conn.accessToken; + let refreshed = false; + let imported = 0; + let windowsWritten = 0; + let newlySettled = 0; + let stopReason: StopReason | null = null; + let retryAfterMs: number | undefined; + let failure: unknown; + const failures: OuraCollectionName[] = []; + + walk: for (const name of OURA_BACKFILL_ORDER) { + const state = progress.collections[name] ?? { before: progress.anchor }; + progress.collections[name] = state; + if (state.done || state.denied) continue; + const floor = floorFor(name, progress); + // The other collections wait for the probes to say where history starts; + // the probes come first, so this is only a slice that ended in them. + if (floor === undefined) break; + + while (!state.done && !state.denied) { + if (state.before <= floor) { + state.done = true; + newlySettled += 1; + settleHistoryStart(progress); + await saveProgress(userId, progress); + break; + } + if (shouldStop() || now() >= deadline) { + stopReason = "budget"; + break walk; + } + if (laneWanted()) { + stopReason = "lane_wanted"; + break walk; + } + const windowDays = WINDOW_DAYS[name] ?? DEFAULT_WINDOW_DAYS; + const shifted = shiftDateKey(state.before, -windowDays); + const startDate = shifted < floor ? floor : shifted; + const window = { startDate, endDate: state.before }; + + let result: { oldestDay?: string; imported: number }; + try { + try { + result = await runWindow(userId, tz, name, window, token); + } catch (err) { + if (!isTokenRefusal(err) || refreshed) throw err; + // A refused token: refresh once for the whole slice, then retry. + refreshed = true; + let next: string | null; + try { + next = await refreshOuraAfterRefusal( + userId, + conn.refreshTokenCiphertext, + ); + } catch (refreshErr) { + await recordOuraSyncFailure(userId, refreshErr); + throw refreshErr; + } + if (!next) throw err; + token = next; + result = await runWindow(userId, tz, name, window, token); + } + } catch (err) { + if ( + isNamedScopeRefusal(err) || + (isTokenRefusal(err) && knownDenied.has(name)) + ) { + // The grant does not cover this collection: a scope note, not a + // failure. The others go on; a reconnect reopens it. + state.denied = true; + newlySettled += 1; + settleHistoryStart(progress); + await saveProgress(userId, progress); + annotate({ meta: { "oura.backfill.denied": name } }); + break; + } + if (err instanceof OuraApiError && err.httpStatus === 429) { + stopReason = "rate_limited"; + retryAfterMs = err.retryAfterMs; + break walk; + } + // One collection's failure stays with it: the slice goes on with the + // next one, and the failed window is read again next slice. + failures.push(name); + failure ??= err; + getEvent()?.addWarning( + `oura backfill: ${name} ${window.startDate}..${window.endDate} failed: ${err}`, + ); + continue walk; + } + + imported += result.imported; + windowsWritten += 1; + if (result.oldestDay) { + progress.oldestSeen = minDay(progress.oldestSeen, result.oldestDay); + } + state.before = startDate; + state.windows = (state.windows ?? 0) + 1; + await saveProgress(userId, progress); + } + } + + const settled = OURA_BACKFILL_ORDER.every((n) => { + const s = progress.collections[n]; + return s?.done || s?.denied; + }); + const report = { + slice, + stopReason, + failures, + windowsWritten, + imported, + edges: Object.fromEntries( + Object.entries(progress.collections).map(([n, s]) => [ + n, + s!.done ? "done" : s!.denied ? "denied" : s!.before, + ]), + ), + historyStart: progress.historyStart ?? null, + }; + annotate({ meta: { "oura.backfill.slice": report } }); + reportJobProgress({ + oura_backfill_slice: slice, + oura_backfill_windows: windowsWritten, + }); + + if (settled && stopReason === null && failures.length === 0) { + await prisma.user.update({ + where: { id: userId }, + data: { + ouraBackfillCompletedAt: new Date(now()), + ouraBackfillProgress: ouraBackfillProgressJson(progress), + }, + }); + annotate({ + action: { name: "oura.backfill.complete", details: { imported, slice } }, + }); + return { imported, continued: false }; + } + + const progressed = windowsWritten > 0 || newlySettled > 0; + // A rate limit passes on its own: the next slice goes after it, progress + // or not, as Strava's backfill does. Anything else needs progress to go on. + if (progressed || stopReason === "rate_limited") { + const delaySeconds = + stopReason === "rate_limited" + ? Math.max( + INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + retryAfterMs !== undefined + ? Math.ceil(retryAfterMs / 1000) + : OURA_BACKFILL_THROTTLED_DELAY_SECONDS, + ) + : stopReason === "lane_wanted" + ? LANE_YIELD_DELAY_SECONDS + : INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS; + if (await sendNextSlice(userId, slice, delaySeconds)) { + annotate({ + action: { + name: "oura.backfill.continued", + details: { slice, stopReason, windowsWritten, delaySeconds }, + }, + }); + return { imported, continued: true }; + } + } + + // Nothing moved, or no next slice could be sent: the pg-boss retry and, + // after it, the next boot's discovery take it from here. The progress + // note stands, so whatever runs next starts where this one stopped. + throw new Error( + `oura backfill incomplete for user ${userId} (slice ${slice}${stopReason ? `, stopped on ${stopReason}` : ""}${failure ? `: ${failure instanceof Error ? failure.message : String(failure)}` : ""})`, + ); +} + +async function sendNextSlice( + userId: string, + slice: number, + delaySeconds: number, +): Promise { + if (slice + 1 > OURA_BACKFILL_MAX_SLICE) return false; + const boss = getGlobalBoss(); + if (!boss) return false; + const sent = await enqueueIntegrationBackfillAdmission( + boss, + { + kind: "oura-backfill", + data: { userId, enqueuedAt: new Date().toISOString(), slice: slice + 1 }, + }, + delaySeconds, + ); + return Boolean(sent); +} + +/** + * Boot-time discovery: every account with an Oura connection whose backfill + * is not stamped, and every account still holding equivalent-distance rows + * from before the fix. Idempotent across reboots; errors come back in the + * result so the worker boot never fails on a miss. + */ +export async function enqueueBootTimeOuraBackfill( + 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 users = await prisma.user.findMany({ + where: { + ouraBackfillCompletedAt: null, + OR: [ + { ouraAccessTokenEncrypted: { not: null } }, + { measurements: { some: OURA_EQUIVALENT_DISTANCE_WHERE } }, + ], + }, + select: { id: true }, + }); + let enqueued = 0; + let skipped = 0; + for (const { id: userId } of users) { + const payload: OuraBackfillPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send( + OURA_BACKFILL_QUEUE, + payload, + integrationBackfillSourceOptions( + `oura-backfill|${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/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 3bc085fd0..80b1feeb1 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -130,6 +130,7 @@ export const QUEUE_RUNTIME: Readonly> = { "oura-sync": cohortSync("Oura"), "strava-sync": cohortSync("Strava"), "strava-backfill": forwarder, + "oura-backfill": forwarder, // ── Status and insights ──────────────────────────────────────────── "insights-general-status": { diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index afb14ee91..d96055632 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -50,6 +50,12 @@ import { enqueueBootTimeStravaBackfill, type StravaBackfillPayload, } from "@/lib/jobs/strava-backfill"; +import { + OURA_BACKFILL_QUEUE, + OURA_BACKFILL_CONCURRENCY, + enqueueBootTimeOuraBackfill, + type OuraBackfillPayload, +} from "@/lib/jobs/oura-backfill"; import { SLEEP_TIMELINE_BACKFILL_QUEUE, SLEEP_TIMELINE_BACKFILL_CONCURRENCY, @@ -325,6 +331,8 @@ const allQueues = [ // v1.17.0 — Polar + Oura OAuth poll sync. POLAR_SYNC_QUEUE, OURA_SYNC_QUEUE, + // v1.44 — Oura history backfill, sliced through the shared lane. + OURA_BACKFILL_QUEUE, // v1.28.x — Strava OAuth poll sync + self-converging boot backfill. STRAVA_SYNC_QUEUE, STRAVA_BACKFILL_QUEUE, @@ -455,6 +463,11 @@ const queuePolicies: QueuePolicyTable = { reason: "Per-connection full-history backfill; boot-discovery driven and self-converging.", }, + [OURA_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection full-history backfill; boot-discovery driven and self-converging.", + }, [GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE]: { policy: "exclusive", reason: @@ -787,6 +800,23 @@ export async function registerIntegrationSyncQueues( return jobDone({ jobs: jobs.length }); }, ); + // v1.44 — Oura history backfill. The source queue keeps the per-account + // singleton and forwards admitted work to the shared drain, where it runs + // in resumable slices. + await createAndWork( + boss, + OURA_BACKFILL_QUEUE, + { localConcurrency: OURA_BACKFILL_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + await enqueueIntegrationBackfillAdmission(boss, { + kind: "oura-backfill", + data: job.data, + }); + } + return jobDone({ jobs: jobs.length }); + }, + ); return allQueues; } @@ -943,6 +973,29 @@ export async function enqueueIntegrationSyncBootDiscovery(): Promise { ); } + // v1.44 — Oura history backfill. Finds every Oura connection not yet + // backfilled (and every account still holding the retired + // equivalent-distance rows) and enqueues one sliced backfill per account. + try { + const { enqueued, skipped, error } = await enqueueBootTimeOuraBackfill( + bootStaggerSecondsFor("oura-backfill"), + ); + if (error) { + workerLog("error", `[oura-backfill] boot discovery failed: ${error}`); + } else { + workerLog( + "info", + `[oura-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[oura-backfill] boot discovery threw an unexpected error", + err, + ); + } + // v1.18.1 — one-shot lab-biomarker backfill. Finds every user holding an // un-linked live lab reading and enqueues one job each. A completed pass // links every reading, dropping the user from the discovery set. diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index 39fa03847..d90560c84 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -46,6 +46,7 @@ const subsystemSurface = { "integration-watchdog-hourly": { audience: "system" }, "integration-watchdog-daily": { audience: "system" }, "strava-backfill": { audience: "system" }, + "oura-backfill": { audience: "system" }, "oidc-native-handoff-cleanup": { audience: "system" }, // register-status.ts diff --git a/src/lib/oura/__tests__/backfill-progress.test.ts b/src/lib/oura/__tests__/backfill-progress.test.ts new file mode 100644 index 000000000..37c6d71d6 --- /dev/null +++ b/src/lib/oura/__tests__/backfill-progress.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it } from "vitest"; + +import { + ouraBackfillProgressJson, + readOuraBackfillProgress, + withoutDenied, +} from "../backfill-progress"; +import { OURA_BACKFILL_ORDER } from "@/lib/jobs/oura-backfill"; + +describe("Oura backfill progress", () => { + it("round-trips a note through the column", () => { + const progress = { + anchor: "2026-10-01", + oldestSeen: "2024-03-02", + historyStart: "2024-03-02", + distanceRetired: true, + collections: { + activity: { before: "2015-01-01", done: true, windows: 47 }, + heartrate: { before: "2025-06-10", windows: 3 }, + workout: { before: "2026-10-01", denied: true }, + }, + }; + const json = ouraBackfillProgressJson(progress); + expect(readOuraBackfillProgress(json, OURA_BACKFILL_ORDER)).toEqual( + progress, + ); + }); + + it("keeps an empty history start apart from an unknown one", () => { + const none = readOuraBackfillProgress( + { anchor: "2026-10-01", historyStart: null, collections: {} }, + OURA_BACKFILL_ORDER, + ); + expect(none?.historyStart).toBeNull(); + const unknown = readOuraBackfillProgress( + { anchor: "2026-10-01", collections: {} }, + OURA_BACKFILL_ORDER, + ); + expect(unknown).not.toHaveProperty("historyStart"); + }); + + it("drops a malformed entry on its own and keeps the rest", () => { + const read = readOuraBackfillProgress( + { + anchor: "2026-10-01", + collections: { + sleep: { before: "yesterday" }, + activity: { before: "2025-01-01" }, + not_a_collection: { before: "2025-01-01" }, + }, + }, + OURA_BACKFILL_ORDER, + ); + expect(read?.collections).toEqual({ activity: { before: "2025-01-01" } }); + }); + + it("reads no note at all without an anchor", () => { + expect(readOuraBackfillProgress(null, OURA_BACKFILL_ORDER)).toBeNull(); + expect( + readOuraBackfillProgress({ collections: {} }, OURA_BACKFILL_ORDER), + ).toBeNull(); + }); + + it("asks again for what a grant refused, and only that", () => { + const cleared = withoutDenied({ + anchor: "2026-10-01", + collections: { + workout: { before: "2026-10-01", denied: true }, + activity: { before: "2015-01-01", done: true }, + }, + }); + expect(cleared.collections).toEqual({ + workout: { before: "2026-10-01" }, + activity: { before: "2015-01-01", done: true }, + }); + }); + + it("walks the history probes first and heart rate last", () => { + expect(OURA_BACKFILL_ORDER.slice(0, 2)).toEqual(["activity", "sleep"]); + expect(OURA_BACKFILL_ORDER.at(-1)).toBe("heartrate"); + expect(new Set(OURA_BACKFILL_ORDER).size).toBe(OURA_BACKFILL_ORDER.length); + }); +}); diff --git a/src/lib/oura/backfill-progress.ts b/src/lib/oura/backfill-progress.ts new file mode 100644 index 000000000..a5c17ef0d --- /dev/null +++ b/src/lib/oura/backfill-progress.ts @@ -0,0 +1,141 @@ +/** + * The `ouraBackfillProgress` column: how far the Oura history backfill has + * got, carried from one slice to the next. + * + * Each collection is walked from the day the backfill began (`anchor`) back + * toward the start of the account's history, one window at a time. After a + * window is written its start becomes the collection's `before`, and the next + * window ends there, so a slice that stops (its budget, a rate limit, a failed + * window) loses at most the window it was in, and the next slice goes on from + * the last one written instead of from the newest day. A collection that + * reached the start of history is `done`; one the grant does not cover is + * `denied` and is asked for again by the first slice of the next chain. + * + * Where history starts is learned, not assumed: `activity` and `sleep` are + * walked first, back to `OURA_HISTORY_FLOOR`, and the oldest day either + * returned (`oldestSeen`) becomes `historyStart`. Every other collection stops + * a week before it, so an account with two years of history costs two years of + * requests, not eleven. + * + * The column is JSON the code wrote, read leniently: a malformed entry is + * dropped on its own and that collection starts over from the anchor, which + * is safe because every write is keyed. + */ +import { z } from "zod"; + +import type { OuraCollectionName } from "./scope"; + +/** The oldest day any Oura ring could have recorded (the first ring shipped in 2015). */ +export const OURA_HISTORY_FLOOR = "2015-01-01"; + +export interface OuraBackfillCollectionState { + /** Upper bound (YYYY-MM-DD) of the next window to read. */ + before: string; + done?: boolean; + denied?: boolean; + /** Windows written across every slice, for the record only. */ + windows?: number; +} + +export interface OuraBackfillProgress { + anchor: string; + /** Oldest day `activity` or `sleep` returned so far. */ + oldestSeen?: string; + /** + * Where the account's history starts, once `activity` and `sleep` are both + * walked to the floor; `null` when they found nothing at all. + */ + historyStart?: string | null; + /** The legacy equivalent-distance rows are retired. */ + distanceRetired?: boolean; + collections: Partial>; +} + +const day = z.string().regex(/^\d{4}-\d{2}-\d{2}$/); + +const stateSchema = z.object({ + before: day, + done: z.boolean().optional().catch(undefined), + denied: z.boolean().optional().catch(undefined), + windows: z.number().int().min(0).optional().catch(undefined), +}); + +/** Read the column; null when there is no usable progress yet. */ +export function readOuraBackfillProgress( + value: unknown, + known: readonly OuraCollectionName[], +): OuraBackfillProgress | null { + if (value === null || typeof value !== "object") return null; + const raw = value as Record; + const anchor = day.safeParse(raw.anchor); + if (!anchor.success) return null; + const collections: OuraBackfillProgress["collections"] = {}; + const rawCollections = + raw.collections && typeof raw.collections === "object" + ? (raw.collections as Record) + : {}; + for (const name of known) { + const parsed = stateSchema.safeParse(rawCollections[name]); + if (parsed.success) { + const s = parsed.data; + collections[name] = { + before: s.before, + ...(s.done ? { done: true } : {}), + ...(s.denied ? { denied: true } : {}), + ...(s.windows !== undefined ? { windows: s.windows } : {}), + }; + } + } + const oldestSeen = day.safeParse(raw.oldestSeen); + const historyStart = + raw.historyStart === null ? null : day.safeParse(raw.historyStart); + return { + anchor: anchor.data, + ...(oldestSeen.success ? { oldestSeen: oldestSeen.data } : {}), + ...(historyStart === null + ? { historyStart: null } + : historyStart.success + ? { historyStart: historyStart.data } + : {}), + ...(raw.distanceRetired === true ? { distanceRetired: true } : {}), + collections, + }; +} + +/** The column's value for a progress note. */ +export function ouraBackfillProgressJson(progress: OuraBackfillProgress) { + return { + anchor: progress.anchor, + ...(progress.oldestSeen ? { oldestSeen: progress.oldestSeen } : {}), + ...(progress.historyStart !== undefined + ? { historyStart: progress.historyStart } + : {}), + ...(progress.distanceRetired ? { distanceRetired: true } : {}), + collections: Object.fromEntries( + Object.entries(progress.collections).map(([name, s]) => [ + name, + { + before: s!.before, + ...(s!.done ? { done: true } : {}), + ...(s!.denied ? { denied: true } : {}), + ...(s!.windows !== undefined ? { windows: s!.windows } : {}), + }, + ]), + ), + }; +} + +/** The progress with every `denied` mark cleared, for a fresh grant. */ +export function withoutDenied( + progress: OuraBackfillProgress, +): OuraBackfillProgress { + return { + ...progress, + collections: Object.fromEntries( + Object.entries(progress.collections).map(([name, s]) => { + const { denied: _denied, ...rest } = s!; + return [name, rest]; + }), + ), + }; +} diff --git a/src/lib/oura/retire-distance.ts b/src/lib/oura/retire-distance.ts new file mode 100644 index 000000000..c79c96dfa --- /dev/null +++ b/src/lib/oura/retire-distance.ts @@ -0,0 +1,72 @@ +/** + * Retire the `WALKING_RUNNING_DISTANCE` rows earlier Oura syncs wrote from + * `equivalent_walking_distance`. + * + * That field is an activity equivalent (the distance a walk of the day's + * effort would have covered), not distance travelled, so the rows were wrong + * from the start and the sync no longer writes them (`mapDailyActivity`). + * What is already stored is soft-deleted here, in bounded batches, and the + * touched days are refolded through the shared mutation tail so the rollups + * stop counting them, the same way `retireOuraSleepRecords` retires a + * rejected sleep period. + * + * Only rows the activity mapper keyed (`activity::distance`) of source + * OURA match; another source's distance, and any OURA row of another shape, + * stay. Idempotent: a second run finds nothing live and returns 0. The Oura + * backfill runs it before anything else on every connected account, and the + * backfill's discovery also offers an account that still holds such rows + * after its Oura connection was removed. + */ +import { prisma } from "@/lib/db"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { annotate } from "@/lib/logging/context"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import type { Prisma } from "@/generated/prisma/client"; + +/** Rows per batch: one bounded read, one update, one refold. */ +export const OURA_DISTANCE_RETIRE_BATCH = 500; + +/** The rows the activity mapper wrote from the equivalent distance. */ +export const OURA_EQUIVALENT_DISTANCE_WHERE = { + source: "OURA", + type: "WALKING_RUNNING_DISTANCE", + deletedAt: null, + externalId: { startsWith: "activity:", endsWith: ":distance" }, +} as const satisfies Prisma.MeasurementWhereInput; + +/** + * Soft-delete every live equivalent-distance row of one account. Returns how + * many rows were retired. Throws on a database failure: the caller decides + * whether that holds its own progress back. + */ +export async function retireOuraEquivalentDistance( + userId: string, +): Promise { + let retired = 0; + for (;;) { + const rows = await prisma.measurement.findMany({ + where: { userId, ...OURA_EQUIVALENT_DISTANCE_WHERE }, + select: { id: true, type: true, measuredAt: true }, + orderBy: { measuredAt: "asc" }, + take: OURA_DISTANCE_RETIRE_BATCH, + }); + if (rows.length === 0) break; + const res = await prisma.measurement.updateMany({ + where: { userId, id: { in: rows.map((r) => r.id) }, deletedAt: null }, + data: { deletedAt: new Date() }, + }); + retired += res.count; + await afterMeasurementMutation(userId, rows, { + source: "sync", + label: "oura", + }); + // A batch the update found already retired (a concurrent run) still + // leaves the read empty next time round, so the loop ends. + if (rows.length < OURA_DISTANCE_RETIRE_BATCH) break; + } + if (retired > 0) { + invalidateUserMeasurements(userId); + annotate({ meta: { oura_equivalent_distance_retired: retired } }); + } + return retired; +} diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index 8c345c0ce..c205e05e5 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -436,7 +436,7 @@ export function ouraCollectionReader( * token and found it valid, the grant just does not cover this collection. A * refresh cannot widen a grant, so none is spent on it. */ -function isNamedScopeRefusal(err: unknown): boolean { +export function isNamedScopeRefusal(err: unknown): boolean { return ( err instanceof OuraApiError && err.httpStatus === 401 && @@ -449,7 +449,7 @@ function isNamedScopeRefusal(err: unknown): boolean { * no body to tell. This is what the one refresh is for. A 403 is neither: Oura * answers it for a lapsed membership, which no refresh or reconnect fixes. */ -function isTokenRefusal(err: unknown): boolean { +export function isTokenRefusal(err: unknown): boolean { return ( err instanceof OuraApiError && err.httpStatus === 401 && diff --git a/tests/integration/oura-backfill.test.ts b/tests/integration/oura-backfill.test.ts new file mode 100644 index 000000000..8dd616149 --- /dev/null +++ b/tests/integration/oura-backfill.test.ts @@ -0,0 +1,502 @@ +/** + * The Oura history backfill against the real schema. + * + * Oura's API is replaced at the `safeFetch` seam by a generator that builds + * each page on request: a year and a half of daily activity, one workout, and + * a heart-rate reading every five minutes over the same span (some 160 000 + * samples, in pages of 1 000). The backfill runs slice after slice, each one + * cut short by its budget, until it stamps its marker. What is pinned: + * + * - it reaches the end: every heart-rate sample is a PULSE row, every day a + * steps row, the workout a Workout row, the marker is stamped, and the + * legacy equivalent-distance rows are retired with their rollup day; + * - it resumes: no window is read twice across slices, apart from the one + * window a 429 interrupted, and that slice sends the next one after the + * Retry-After the 429 named; + * - it does not grow with the history: the collected heap after the first + * quarter of heart-rate pages stays inside a budget a collect-everything + * walk of the same pages passes several times over; + * - discovery offers an existing connection without a reconnect, and an + * account that kept only the legacy rows, and nothing already stamped. + */ +import v8 from "node:v8"; +import vm from "node:vm"; + +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER = "oura-backfill-owner"; +const DAY_MS = 24 * 60 * 60 * 1000; +/** The day the backfill begins, and the end of the generated history. */ +const ANCHOR = "2026-10-01"; +const NOW = Date.parse(`${ANCHOR}T12:00:00Z`); +/** The first day of the generated history. */ +const HISTORY_FROM = "2025-04-01"; +const HR_FROM = Date.parse(`${HISTORY_FROM}T00:00:00Z`); +const HR_STEP = 5 * 60_000; +const HR_PAGE = 1000; +const HR_TOTAL = Math.floor((NOW - HR_FROM) / HR_STEP); + +const GROWTH_BUDGET_BYTES = 24 * 1024 * 1024; + +const fake = vi.hoisted(() => ({ + requests: [] as string[], + heartRatePages: 0, + /** Answer the next heart-rate page with a 429 once this many have been served. */ + rateLimitAtPage: -1, + onPage: null as (() => void) | null, + /** Collections answered with a 401 naming their scope. */ + denied: new Set(), + sends: [] as Array<{ queue: string; payload: unknown; options: unknown }>, +})); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => ({ + send: async (queue: string, payload: unknown, options: unknown) => { + fake.sends.push({ queue, payload, options }); + return `job-${fake.sends.length}`; + }, + }), +})); + +function dayKey(ms: number): string { + return new Date(ms).toISOString().slice(0, 10); +} + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + const json = (body: unknown, status = 200, headers = {}) => + new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json", ...headers }, + }); + return { + ...actual, + // A plain function, not `vi.fn`: a mock records every result, and the + // results here are whole pages. + safeFetch: async (url: string) => { + const u = new URL(url); + const p = u.searchParams; + const collection = u.pathname.split("/").pop()!; + const token = p.get("next_token"); + fake.requests.push( + `${collection}|${p.get("start_date") ?? p.get("start_datetime")}|${p.get("end_date") ?? p.get("end_datetime")}|${token ?? ""}`, + ); + if (fake.denied.has(collection)) { + return json( + { detail: `Token is not authorized access ${collection} scope.` }, + 401, + ); + } + if (collection === "heartrate") { + if (fake.rateLimitAtPage === fake.heartRatePages) { + fake.rateLimitAtPage = -1; + return json({ detail: "Too many requests" }, 429, { + "retry-after": "600", + }); + } + const start = Date.parse(p.get("start_datetime")!); + const end = Date.parse(p.get("end_datetime")!); + const first = Math.max(0, Math.ceil((start - HR_FROM) / HR_STEP)); + const last = Math.min(HR_TOTAL, Math.ceil((end - HR_FROM) / HR_STEP)); + const from = first + Number(token ?? 0) * HR_PAGE; + const to = Math.min(last, from + HR_PAGE); + const data = []; + for (let i = from; i < to; i++) { + data.push({ + bpm: 52 + (i % 40), + source: "awake", + timestamp: new Date(HR_FROM + i * HR_STEP).toISOString(), + }); + } + fake.heartRatePages++; + fake.onPage?.(); + return json({ + data, + next_token: to < last ? String(Number(token ?? 0) + 1) : null, + }); + } + if (collection === "daily_activity") { + const from = + p.get("start_date")! < HISTORY_FROM + ? HISTORY_FROM + : p.get("start_date")!; + const to = p.get("end_date")! > ANCHOR ? ANCHOR : p.get("end_date")!; + const data = []; + for ( + let t = Date.parse(`${from}T00:00:00Z`); + dayKey(t) <= to; + t += DAY_MS + ) { + data.push({ + id: `act-${dayKey(t)}`, + day: dayKey(t), + steps: 5000, + active_calories: 300, + equivalent_walking_distance: 4000, + }); + } + return json({ data, next_token: null }); + } + if (collection === "workout") { + const day = "2026-02-14"; + const inRange = + p.get("start_date")! <= day && day <= p.get("end_date")!; + return json({ + data: inRange + ? [ + { + id: "oura-workout-1", + activity: "cycling", + calories: 410, + distance: 15000, + intensity: "moderate", + source: "manual", + day, + start_datetime: `${day}T08:00:00+00:00`, + end_datetime: `${day}T09:00:00+00:00`, + }, + ] + : [], + next_token: null, + }); + } + return json({ data: [], next_token: null }); + }, + }; +}); + +const forceGc = ((): (() => void) => { + v8.setFlagsFromString("--expose-gc"); + const gc = vm.runInNewContext("gc") as () => void; + v8.setFlagsFromString("--no-expose-gc"); + return gc; +})(); + +function liveHeapBytes(): number { + forceGc(); + forceGc(); + return process.memoryUsage().heapUsed; +} + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + fake.requests = []; + fake.heartRatePages = 0; + fake.rateLimitAtPage = -1; + fake.sends = []; + fake.denied = new Set(); + const { encrypt } = await import("@/lib/crypto"); + await prisma.user.create({ + data: { + id: USER, + username: USER, + timezone: "UTC", + ouraAccessTokenEncrypted: encrypt("access"), + ouraRefreshTokenEncrypted: encrypt("refresh"), + }, + }); +}, 120_000); + +/** Seed the rows an earlier sync wrote from the equivalent distance. */ +async function seedLegacyDistance(userId: string): Promise { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + { + userId, + type: "WALKING_RUNNING_DISTANCE", + source: "OURA", + value: 4000, + unit: "m", + measuredAt: new Date("2026-09-01T12:00:00Z"), + externalId: "activity:2026-09-01:distance", + }, + { + userId, + type: "WALKING_RUNNING_DISTANCE", + source: "OURA", + value: 5200, + unit: "m", + measuredAt: new Date("2026-09-02T12:00:00Z"), + externalId: "activity:2026-09-02:distance", + }, + // Another source's distance on the same day stays. + { + userId, + type: "WALKING_RUNNING_DISTANCE", + source: "APPLE_HEALTH", + value: 3100, + unit: "m", + measuredAt: new Date("2026-09-01T18:00:00Z"), + externalId: "hk-dist-1", + }, + ], + }); +} + +describe("retireOuraEquivalentDistance", () => { + it("soft-deletes only the OURA equivalent-distance rows and refolds the day", async () => { + const prisma = getPrismaClient(); + await seedLegacyDistance(USER); + const { recomputeBucketsForMeasurement } = + await import("@/lib/rollups/measurement-rollups"); + await recomputeBucketsForMeasurement( + USER, + "WALKING_RUNNING_DISTANCE", + new Date("2026-09-01T12:00:00Z"), + ); + const ouraDay = () => + prisma.measurementRollup.findFirst({ + where: { + userId: USER, + type: "WALKING_RUNNING_DISTANCE", + granularity: "DAY", + source: "OURA", + bucketStart: new Date("2026-09-01T00:00:00Z"), + }, + }); + expect((await ouraDay())?.count).toBe(1); + + const { retireOuraEquivalentDistance } = + await import("@/lib/oura/retire-distance"); + expect(await retireOuraEquivalentDistance(USER)).toBe(2); + expect(await retireOuraEquivalentDistance(USER)).toBe(0); + + const live = await prisma.measurement.findMany({ + where: { userId: USER, deletedAt: null }, + select: { source: true, externalId: true }, + }); + expect(live).toEqual([{ source: "APPLE_HEALTH", externalId: "hk-dist-1" }]); + const after = await ouraDay(); + expect(after?.count ?? 0).toBe(0); + }); +}); + +describe("Oura history backfill", () => { + it("walks the whole history in resumable slices with bounded memory", async () => { + const prisma = getPrismaClient(); + await seedLegacyDistance(USER); + const { runOuraBackfillForUser } = await import("@/lib/jobs/oura-backfill"); + + // A rate limit in the middle of the heart-rate walk. + fake.rateLimitAtPage = 40; + + let clock = NOW; + const now = () => (clock += 1000); + const quarter = Math.ceil(HR_TOTAL / HR_PAGE / 4); + let anchor: number | null = null; + let peak = 0; + let samples = 0; + fake.onPage = () => { + if (fake.heartRatePages % 5 !== 0) return; + const live = liveHeapBytes(); + if (anchor === null) { + if (fake.heartRatePages >= quarter) anchor = peak = live; + return; + } + samples++; + peak = Math.max(peak, live); + }; + + let slice = 0; + let continued = true; + let slices = 0; + try { + while (continued) { + const run = await runOuraBackfillForUser(USER, () => false, slice, { + sliceMs: 40_000, + now, + }); + continued = run.continued; + slice += 1; + slices += 1; + expect(slices).toBeLessThan(200); + } + } finally { + fake.onPage = null; + } + process.stderr.write( + `[oura-backfill] slices=${slices} hrPages=${fake.heartRatePages} growth=${((peak - (anchor ?? 0)) / 1024 / 1024).toFixed(1)} MB over ${samples} samples\n`, + ); + + // It ran in many slices, and finished. + expect(slices).toBeGreaterThan(5); + const user = await prisma.user.findUniqueOrThrow({ + where: { id: USER }, + select: { ouraBackfillCompletedAt: true, ouraBackfillProgress: true }, + }); + expect(user.ouraBackfillCompletedAt).not.toBeNull(); + const progress = user.ouraBackfillProgress as { + historyStart: string; + distanceRetired: boolean; + }; + expect(progress.historyStart).toBe(HISTORY_FROM); + expect(progress.distanceRetired).toBe(true); + + // Every heart-rate sample, every day, the workout. + expect( + await prisma.measurement.count({ + where: { userId: USER, type: "PULSE", source: "OURA" }, + }), + ).toBe(HR_TOTAL); + const days = + (Date.parse(`${ANCHOR}T00:00:00Z`) - + Date.parse(`${HISTORY_FROM}T00:00:00Z`)) / + DAY_MS + + 1; + expect( + await prisma.measurement.count({ + where: { userId: USER, type: "ACTIVITY_STEPS", source: "OURA" }, + }), + ).toBe(days); + const workouts = await prisma.workout.findMany({ + where: { userId: USER, source: "OURA" }, + select: { externalId: true, sportType: true, totalDistanceM: true }, + }); + expect(workouts).toEqual([ + { + externalId: "oura-workout-1", + sportType: "cycling", + totalDistanceM: 15000, + }, + ]); + // No equivalent distance written, and the legacy rows retired. + expect( + await prisma.measurement.count({ + where: { + userId: USER, + source: "OURA", + type: "WALKING_RUNNING_DISTANCE", + deletedAt: null, + }, + }), + ).toBe(0); + + // Resumed, never restarted: every first page of a window was asked for + // once, apart from the heart-rate window the 429 interrupted. + const firstPages = fake.requests.filter((r) => r.endsWith("|")); + const repeated = firstPages.filter((r, i) => firstPages.indexOf(r) !== i); + expect(repeated).toHaveLength(1); + expect(repeated[0]).toMatch(/^heartrate\|/); + // The slice the 429 ended sent the next one after the wait it named. + const delays = fake.sends.map( + (s) => (s.options as { startAfter?: number }).startAfter ?? 0, + ); + expect(delays).toContain(600); + + // Bounded: the heap after the first quarter of the stream stays flat. + expect(samples).toBeGreaterThan(5); + expect(peak - anchor!).toBeLessThan(GROWTH_BUDGET_BYTES); + }, 900_000); + + it("leaves an existing connection's history to discovery, no reconnect needed", async () => { + const prisma = getPrismaClient(); + // An account whose Oura connection is gone but whose legacy rows stay. + await prisma.user.create({ + data: { id: "oura-legacy-only", username: "oura-legacy-only" }, + }); + await seedLegacyDistance("oura-legacy-only"); + // An account already backfilled. + await prisma.user.create({ + data: { + id: "oura-done", + username: "oura-done", + ouraAccessTokenEncrypted: "x", + ouraRefreshTokenEncrypted: "y", + ouraBackfillCompletedAt: new Date(), + }, + }); + const { enqueueBootTimeOuraBackfill, runOuraBackfillForUser } = + await import("@/lib/jobs/oura-backfill"); + + const result = await enqueueBootTimeOuraBackfill(30); + expect(result).toEqual({ enqueued: 2, skipped: 0, error: null }); + expect( + fake.sends.map((s) => (s.payload as { userId: string }).userId).sort(), + ).toEqual(["oura-backfill-owner", "oura-legacy-only"]); + + // The legacy-only account retires its rows and is stamped at once. + await runOuraBackfillForUser("oura-legacy-only"); + const legacy = await prisma.user.findUniqueOrThrow({ + where: { id: "oura-legacy-only" }, + select: { ouraBackfillCompletedAt: true }, + }); + expect(legacy.ouraBackfillCompletedAt).not.toBeNull(); + expect( + await prisma.measurement.count({ + where: { userId: "oura-legacy-only", source: "OURA", deletedAt: null }, + }), + ).toBe(0); + }); + + it("finishes past a collection the grant refuses, and asks again after a reconnect", async () => { + const prisma = getPrismaClient(); + const { OURA_BACKFILL_ORDER, runOuraBackfillForUser } = + await import("@/lib/jobs/oura-backfill"); + // Everything but workouts already walked. + await prisma.user.update({ + where: { id: USER }, + data: { + ouraBackfillProgress: { + anchor: ANCHOR, + historyStart: "2026-01-01", + distanceRetired: true, + collections: Object.fromEntries( + OURA_BACKFILL_ORDER.filter((n) => n !== "workout").map((n) => [ + n, + { before: "2015-01-01", done: true }, + ]), + ), + }, + }, + }); + fake.denied.add("workout"); + + expect(await runOuraBackfillForUser(USER)).toEqual({ + imported: 0, + continued: false, + }); + let user = await prisma.user.findUniqueOrThrow({ + where: { id: USER }, + select: { ouraBackfillCompletedAt: true, ouraBackfillProgress: true }, + }); + expect(user.ouraBackfillCompletedAt).not.toBeNull(); + expect( + (user.ouraBackfillProgress as { collections: Record }) + .collections.workout, + ).toEqual({ before: ANCHOR, denied: true }); + // One request found it out; nothing was recorded as a failure. + expect(fake.requests.filter((r) => r.startsWith("workout|"))).toHaveLength( + 1, + ); + expect(await prisma.integrationStatus.count()).toBe(0); + + // A reconnect that grants the scope clears the stamp; the next chain + // reads workouts and nothing else. + fake.denied.clear(); + fake.requests = []; + await prisma.user.update({ + where: { id: USER }, + data: { ouraBackfillCompletedAt: null }, + }); + await runOuraBackfillForUser(USER); + user = await prisma.user.findUniqueOrThrow({ + where: { id: USER }, + select: { ouraBackfillCompletedAt: true, ouraBackfillProgress: true }, + }); + expect(user.ouraBackfillCompletedAt).not.toBeNull(); + expect(new Set(fake.requests.map((r) => r.split("|")[0]))).toEqual( + new Set(["workout"]), + ); + expect( + await prisma.workout.count({ where: { userId: USER, source: "OURA" } }), + ).toBe(1); + }); +});