diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 6ee366015..4e57af7e3 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -79547,6 +79547,8 @@ components: - vo2max - cardio_age - resilience + - workout + - heartrate hasToken: description: Nightscout only. type: boolean diff --git a/messages/de.json b/messages/de.json index 5096b7c90..7f0fe9c78 100644 --- a/messages/de.json +++ b/messages/de.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "VO2max", "cardio_age": "kardiovaskuläres Alter", - "resilience": "Resilienz" + "resilience": "Resilienz", + "workout": "Trainings", + "heartrate": "Herzfrequenz" } } } diff --git a/messages/en.json b/messages/en.json index 37ff91d18..178ac403d 100644 --- a/messages/en.json +++ b/messages/en.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "VO2 max", "cardio_age": "cardiovascular age", - "resilience": "resilience" + "resilience": "resilience", + "workout": "workouts", + "heartrate": "heart rate" } } } diff --git a/messages/es.json b/messages/es.json index 4b1a9e65b..dca997506 100644 --- a/messages/es.json +++ b/messages/es.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "VO2 máx.", "cardio_age": "edad cardiovascular", - "resilience": "resiliencia" + "resilience": "resiliencia", + "workout": "entrenamientos", + "heartrate": "frecuencia cardiaca" } } } diff --git a/messages/fr.json b/messages/fr.json index ea046e3a4..7edc2acbf 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "la VO2 max", "cardio_age": "l’âge cardiovasculaire", - "resilience": "la résilience" + "resilience": "la résilience", + "workout": "les entraînements", + "heartrate": "la fréquence cardiaque" } } } diff --git a/messages/it.json b/messages/it.json index 47b448231..0c2842998 100644 --- a/messages/it.json +++ b/messages/it.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "VO2 max", "cardio_age": "età cardiovascolare", - "resilience": "resilienza" + "resilience": "resilienza", + "workout": "allenamenti", + "heartrate": "frequenza cardiaca" } } } diff --git a/messages/ko.json b/messages/ko.json index a0515cf73..a5a7b7d09 100644 --- a/messages/ko.json +++ b/messages/ko.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "VO2 max", "cardio_age": "심혈관 나이", - "resilience": "회복탄력성" + "resilience": "회복탄력성", + "workout": "운동", + "heartrate": "심박수" } } } diff --git a/messages/nl.json b/messages/nl.json index a5e20c7d3..511ee00d6 100644 --- a/messages/nl.json +++ b/messages/nl.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "VO2 max", "cardio_age": "cardiovasculaire leeftijd", - "resilience": "veerkracht" + "resilience": "veerkracht", + "workout": "trainingen", + "heartrate": "hartslag" } } } diff --git a/messages/pl.json b/messages/pl.json index 394f3ccbf..5f8e23289 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -6672,7 +6672,9 @@ "items": { "vo2max": "VO2 max", "cardio_age": "wiek sercowo-naczyniowy", - "resilience": "odporność" + "resilience": "odporność", + "workout": "treningi", + "heartrate": "tętno" } } } 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-plausibility-guard.test.ts b/src/__tests__/measurement-write-plausibility-guard.test.ts index 9b3cf83e0..e334df976 100644 --- a/src/__tests__/measurement-write-plausibility-guard.test.ts +++ b/src/__tests__/measurement-write-plausibility-guard.test.ts @@ -62,6 +62,7 @@ const WRITERS: Record = { "lib/fitbit/sync-core.ts": { kind: "gate" }, "lib/google-health/sync-core.ts": { kind: "gate" }, "lib/nightscout/sync.ts": { kind: "gate" }, + "lib/oura/sync-heart-rate.ts": { kind: "gate" }, // The shared identity seam. Polar, Oura, WHOOP, the Withings sleep leg, // the Apple export importer and the batch endpoint all write through it, // so the gate here is what makes a new provider fail closed. diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index da045a6ed..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": @@ -61,6 +63,8 @@ const WRITE_SITES: Record = { "refolds the touched days and re-warms the touched types at the end of the upsert", "lib/nightscout/sync.ts": "refolds the touched days and re-warms the touched types at the end of the sync", + "lib/oura/sync-heart-rate.ts": + "collects the days each heart-rate page touched; refolds them and re-warms the type once per window", "lib/withings/sync.ts": "refolds the touched days and re-warms the touched types at the end of the sync", "lib/withings/sync-activity.ts": diff --git a/src/__tests__/workout-sport-provenance-guard.test.ts b/src/__tests__/workout-sport-provenance-guard.test.ts index f135279b1..9eda369d3 100644 --- a/src/__tests__/workout-sport-provenance-guard.test.ts +++ b/src/__tests__/workout-sport-provenance-guard.test.ts @@ -116,6 +116,15 @@ const PROVIDER_INGESTS: ProviderIngest[] = [ writeFile: "lib/import/health-connect/import.ts", metadataKey: "healthConnectExerciseType", }, + { + provider: "Oura", + mapper: "mapOuraSportType", + mapperFile: "lib/oura/sport-map.ts", + captureFile: "lib/oura/sync-workouts.ts", + captureField: "w.activity", + writeFile: "lib/oura/sync-workouts.ts", + metadataKey: "ouraActivity", + }, { provider: "Fitbit", mapper: "mapFitbitSportType", diff --git a/src/app/api/oura/callback/__tests__/route.test.ts b/src/app/api/oura/callback/__tests__/route.test.ts index 23e5b7b5c..342c6d1df 100644 --- a/src/app/api/oura/callback/__tests__/route.test.ts +++ b/src/app/api/oura/callback/__tests__/route.test.ts @@ -218,12 +218,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 21d3c60d6..253ac4f60 100644 --- a/src/app/api/oura/callback/route.ts +++ b/src/app/api/oura/callback/route.ts @@ -18,6 +18,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"; /** @@ -143,6 +148,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, }, }); @@ -165,6 +174,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/components/settings/integrations/integration-health-copy.ts b/src/components/settings/integrations/integration-health-copy.ts index f9340dd4e..17e8f8cfa 100644 --- a/src/components/settings/integrations/integration-health-copy.ts +++ b/src/components/settings/integrations/integration-health-copy.ts @@ -130,6 +130,10 @@ export function unavailableItemLabel(item: string, t: Translate): string { return t("settings.integrations.oura.scope.items.cardio_age"); case "resilience": return t("settings.integrations.oura.scope.items.resilience"); + case "workout": + return t("settings.integrations.oura.scope.items.workout"); + case "heartrate": + return t("settings.integrations.oura.scope.items.heartrate"); case "activity": return t("settings.integrationCard.items.activity"); case "sleep": 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__/oura-backfill.test.ts b/src/lib/jobs/__tests__/oura-backfill.test.ts new file mode 100644 index 000000000..42a9dfc40 --- /dev/null +++ b/src/lib/jobs/__tests__/oura-backfill.test.ts @@ -0,0 +1,301 @@ +/** + * The Oura backfill's slice logic with the provider and the database stubbed: + * where a slice stops, what it saves, where the next one starts, and how long + * it waits. The real walk against Postgres is + * `tests/integration/oura-backfill.test.ts`. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const h = vi.hoisted(() => ({ + stored: null as unknown, + completedAt: null as Date | null, + sends: [] as Array<{ payload: unknown; startAfter?: number }>, + windows: [] as string[], + /** Per collection: what a window read does. */ + behave: {} as Record< + string, + (w: { startDate: string; endDate: string }) => Promise + >, +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + findUnique: vi.fn(async () => ({ + ouraBackfillProgress: h.stored, + ouraBackfillCompletedAt: h.completedAt, + })), + update: vi.fn( + async ({ + data, + }: { + data: { + ouraBackfillProgress?: unknown; + ouraBackfillCompletedAt?: Date; + }; + }) => { + if (data.ouraBackfillProgress !== undefined) { + h.stored = JSON.parse(JSON.stringify(data.ouraBackfillProgress)); + } + if (data.ouraBackfillCompletedAt) { + h.completedAt = data.ouraBackfillCompletedAt; + } + }, + ), + }, + }, +})); +vi.mock("@/lib/oura/retire-distance", () => ({ + OURA_EQUIVALENT_DISTANCE_WHERE: {}, + retireOuraEquivalentDistance: vi.fn(async () => 0), +})); +vi.mock("@/lib/oura/credentials", () => ({ + getOuraConnection: vi.fn(async () => ({ + accessToken: "acc", + refreshTokenCiphertext: "enc", + scopeDenied: [], + })), +})); +vi.mock("@/lib/oura/oauth-descriptor", () => ({ + refreshOuraAfterRefusal: vi.fn(async () => "fresh"), +})); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: vi.fn(async () => false), +})); +vi.mock("@/lib/tz/resolver", () => ({ + resolveUserTimezone: vi.fn(async () => "UTC"), +})); +vi.mock("@/lib/sleep/sweep-stale-segments", () => ({ + sweepStaleSleepSegments: vi.fn(async () => {}), +})); +vi.mock("@/lib/oura/sync-heart-rate", async (importOriginal) => ({ + ...(await importOriginal()), + refoldOuraTouchedDays: vi.fn(async () => {}), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => ({ + send: async ( + _queue: string, + payload: unknown, + options: { startAfter?: number }, + ) => { + h.sends.push({ payload, startAfter: options.startAfter }); + return "job"; + }, + }), +})); +vi.mock("@/lib/oura/sync", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + ouraCollectionReader: + (name: string, ctx: { query: { startDate: string; endDate: string } }) => + async () => { + h.windows.push(`${name}:${ctx.query.startDate}..${ctx.query.endDate}`); + return (h.behave[name] ?? (async () => []))(ctx.query); + }, + upsertOuraMeasurements: vi.fn( + async (_u: string, readings: unknown[]) => readings.length, + ), + retireOuraSleepRecords: vi.fn(async () => 0), + recordOuraSyncFailure: vi.fn(async () => {}), + }; +}); + +import { OuraApiError } from "@/lib/oura/response-classifier"; +import { OURA_BACKFILL_ORDER, runOuraBackfillForUser } from "../oura-backfill"; + +const T0 = Date.parse("2026-10-01T12:00:00Z"); + +/** Activity on every window from mid-2026: history starts 2026-06-01. */ +function activityFrom(first: string) { + return async (w: { startDate: string; endDate: string }) => + w.endDate < first + ? [] + : [ + { + type: "ACTIVITY_STEPS", + value: 1, + unit: "steps", + measuredAt: new Date( + `${w.startDate < first ? first : w.startDate}T12:00:00Z`, + ), + externalId: `activity:${w.startDate}:steps`, + }, + ]; +} + +/** A clock that moves one second per read. */ +function clock() { + let t = T0; + return () => (t += 1000); +} + +beforeEach(() => { + h.stored = null; + h.completedAt = null; + h.sends = []; + h.windows = []; + h.behave = { activity: activityFrom("2026-06-01") }; +}); + +describe("runOuraBackfillForUser", () => { + it("stops on its budget, saves where each collection got to, and resumes there", async () => { + const first = await runOuraBackfillForUser("u1", () => false, 0, { + sliceMs: 10_000, + now: clock(), + }); + expect(first.continued).toBe(true); + expect(h.sends[0]!.payload).toMatchObject({ + kind: "oura-backfill", + data: { userId: "u1", slice: 1 }, + }); + const saved = h.stored as { + anchor: string; + collections: Record; + }; + expect(saved.anchor).toBe("2026-10-01"); + const lastWindow = h.windows.at(-1)!; + const [name, range] = lastWindow.split(":"); + expect(saved.collections[name!]!.before).toBe(range!.split("..")[0]); + + // The next slice opens with the window right before the saved edge. + h.windows = []; + await runOuraBackfillForUser("u1", () => false, 1, { + sliceMs: 10_000, + now: clock(), + }); + expect(h.windows[0]!.startsWith(`${name}:`)).toBe(true); + expect( + h.windows[0]!.endsWith(`..${saved.collections[name!]!.before}`), + ).toBe(true); + }); + + it("runs to the end and learns where history starts", async () => { + let slice = 0; + let continued = true; + while (continued) { + continued = ( + await runOuraBackfillForUser("u1", () => false, slice++, { + sliceMs: 60_000, + now: clock(), + }) + ).continued; + expect(slice).toBeLessThan(100); + } + expect(h.completedAt).not.toBeNull(); + const saved = h.stored as { + historyStart: string; + collections: Record; + }; + expect(saved.historyStart).toBe("2026-06-01"); + for (const name of OURA_BACKFILL_ORDER) { + expect(saved.collections[name]!.done).toBe(true); + } + // The non-probe collections stop a week before the history start. + expect(saved.collections.heartrate!.before).toBe("2026-05-25"); + expect( + h.windows + .filter((w) => w.startsWith("heartrate:")) + .every((w) => { + const start = w.split(":")[1]!.split("..")[0]!; + return start >= "2026-05-25"; + }), + ).toBe(true); + // No window read twice. + expect(new Set(h.windows).size).toBe(h.windows.length); + }); + + it("waits out the Retry-After a 429 named before the next slice", async () => { + h.behave.activity = async () => { + throw new OuraApiError({ + verb: "fetchDailyActivity", + classification: "transient", + httpStatus: 429, + reason: "http_429", + retryAfterMs: 900_000, + }); + }; + const run = await runOuraBackfillForUser("u1", () => false, 3, { + now: clock(), + }); + expect(run.continued).toBe(true); + expect(h.sends).toHaveLength(1); + expect(h.sends[0]!.startAfter).toBe(900); + expect(h.sends[0]!.payload).toMatchObject({ data: { slice: 4 } }); + }); + + it("marks a collection the grant refuses as denied, finishes, and asks again on the next chain", async () => { + h.behave.workout = async () => { + throw new OuraApiError({ + verb: "fetchWorkouts", + classification: "reauth_required", + httpStatus: 401, + reason: "scope_missing", + missingScope: "workout", + }); + }; + let slice = 0; + let continued = true; + while (continued) { + continued = ( + await runOuraBackfillForUser("u1", () => false, slice++, { + sliceMs: 60_000, + now: clock(), + }) + ).continued; + } + expect(h.completedAt).not.toBeNull(); + const saved = h.stored as { + collections: Record; + }; + expect(saved.collections.workout!.denied).toBe(true); + expect(h.windows.filter((w) => w.startsWith("workout:"))).toHaveLength(1); + + // A connect reopens it; the first slice asks for workouts again and + // nothing else. + h.completedAt = null; + h.windows = []; + delete h.behave.workout; + await runOuraBackfillForUser("u1", () => false, 0, { + sliceMs: 60_000, + now: clock(), + }); + expect(new Set(h.windows.map((w) => w.split(":")[0]))).toEqual( + new Set(["workout"]), + ); + expect(h.completedAt).not.toBeNull(); + }); + + it("goes on with the other collections past a failing one, and throws when nothing moved", async () => { + h.behave.spo2 = async () => { + throw new OuraApiError({ + verb: "fetchDailySpo2", + classification: "transient", + httpStatus: 502, + reason: "http_502", + }); + }; + const run = await runOuraBackfillForUser("u1", () => false, 0, { + sliceMs: 600_000, + now: clock(), + }); + // Everything but spo2 moved, so the chain goes on. + expect(run.continued).toBe(true); + const saved = h.stored as { + collections: Record; + }; + expect(saved.collections.heartrate!.done).toBe(true); + expect(saved.collections.spo2!.done).toBeUndefined(); + + // Now only spo2 is left, and it fails again: no progress, so it throws + // for the pg-boss retry instead of chaining forever. + await expect( + runOuraBackfillForUser("u1", () => false, 1, { + sliceMs: 600_000, + now: clock(), + }), + ).rejects.toThrow(/oura backfill incomplete/); + expect(h.completedAt).toBeNull(); + }); +}); 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/__tests__/client.test.ts b/src/lib/oura/__tests__/client.test.ts index 13535d03c..46d800995 100644 --- a/src/lib/oura/__tests__/client.test.ts +++ b/src/lib/oura/__tests__/client.test.ts @@ -77,7 +77,9 @@ describe("getAuthorizationUrl", () => { it("builds the Oura authorize URL", () => { const url = getAuthorizationUrl("signed", CREDS); expect(url).toContain("cloud.ouraring.com/oauth/authorize"); - expect(url).toContain("scope=daily+spo2+heart_health+stress+personal"); + expect(url).toContain( + "scope=daily+spo2+heart_health+stress+workout+heartrate+personal", + ); expect(url).toContain("state=signed"); }); }); @@ -471,7 +473,7 @@ describe("mapSleep", () => { }); describe("mapDailyActivity", () => { - it("maps steps, active energy, and equivalent walking distance (m)", () => { + it("maps steps and active energy, and never the equivalent walking distance", () => { const a: OuraDailyActivity = { id: "1", day: "2026-06-10", @@ -484,12 +486,12 @@ describe("mapDailyActivity", () => { expect(mapped.find((m) => m.type === "ACTIVE_ENERGY_BURNED")?.value).toBe( 412, ); - const dist = mapped.find((m) => m.type === "WALKING_RUNNING_DISTANCE"); - expect(dist).toMatchObject({ - value: 6543, - unit: "m", - fieldTag: "distance", - }); + // An activity equivalent, not distance travelled: no distance row at all. + expect(mapped.map((m) => m.type).sort()).toEqual([ + "ACTIVE_ENERGY_BURNED", + "ACTIVITY_STEPS", + ]); + expect(mapped.some((m) => m.fieldTag === "distance")).toBe(false); }); }); diff --git a/src/lib/oura/__tests__/scope.test.ts b/src/lib/oura/__tests__/scope.test.ts index caf2bdfab..0df17a1a4 100644 --- a/src/lib/oura/__tests__/scope.test.ts +++ b/src/lib/oura/__tests__/scope.test.ts @@ -30,6 +30,8 @@ describe("Oura scope map", () => { "spo2", "heart_health", "stress", + "workout", + "heartrate", "personal", ]); }); @@ -40,6 +42,8 @@ describe("Oura scope map", () => { "vo2max", "cardio_age", "resilience", + "workout", + "heartrate", ]); expect(ouraScopeReconnectList(["resilience", "readiness", "spo2"])).toEqual( ["spo2", "resilience"], @@ -65,11 +69,18 @@ describe("granted scopes", () => { ouraCollectionsMissingFromGrant( "extapi:daily extapi:personal extapi:spo2 extapi:heart_health", ), - ).toEqual(["resilience"]); + ).toEqual(["resilience", "workout", "heartrate"]); + // A grant from before workouts and heart rate were read: those two are + // the scope note, nothing else. expect( ouraCollectionsMissingFromGrant( "daily personal spo2 heart_health stress", ), + ).toEqual(["workout", "heartrate"]); + expect( + ouraCollectionsMissingFromGrant( + "daily personal spo2 heart_health stress workout heartrate", + ), ).toEqual([]); }); }); diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index 63616a85d..cd2de619b 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/src/lib/oura/__tests__/sync.test.ts @@ -21,6 +21,8 @@ const { fetchCardioAgeMock, fetchResilienceMock, fetchCyclePhasesMock, + fetchWorkoutsMock, + heartRatePagesMock, refreshMock, createManyAndReturnMock, updateMock, @@ -48,6 +50,8 @@ const { fetchCardioAgeMock: vi.fn(), fetchResilienceMock: vi.fn(), fetchCyclePhasesMock: vi.fn(), + fetchWorkoutsMock: vi.fn(), + heartRatePagesMock: vi.fn(), refreshMock: vi.fn(), createManyAndReturnMock: vi.fn(), updateMock: vi.fn(), @@ -142,6 +146,8 @@ vi.mock("../client", async (importOriginal) => { fetchCardiovascularAge: fetchCardioAgeMock, fetchResilience: fetchResilienceMock, fetchDailyCyclePhases: fetchCyclePhasesMock, + fetchWorkouts: fetchWorkoutsMock, + forEachHeartRatePage: heartRatePagesMock, refreshAccessToken: refreshMock, }; }); @@ -158,6 +164,13 @@ import { isSyncFailureRecorded } from "@/lib/integrations/status"; const DAY_MS = 24 * 60 * 60 * 1000; +/** The day-window watermark reads (the heart-rate window's carries a type). */ +function dayWatermarkReads() { + return findFirstMock.mock.calls.filter( + ([args]) => !(args as { where: { type?: string } }).where.type, + ); +} + const CONN = { accessToken: "acc", refreshTokenCiphertext: "enc(ref)", @@ -180,6 +193,10 @@ beforeEach(() => { fetchCardioAgeMock.mockReset().mockResolvedValue([]); fetchResilienceMock.mockReset().mockResolvedValue([]); fetchCyclePhasesMock.mockReset().mockResolvedValue([]); + fetchWorkoutsMock.mockReset().mockResolvedValue([]); + heartRatePagesMock + .mockReset() + .mockResolvedValue({ pages: 1, truncated: false }); refreshMock.mockReset(); scopeDeniedMock.mockReset().mockResolvedValue(undefined); transactionMock @@ -902,9 +919,10 @@ describe("syncUserOura — window resolution", () => { await syncUserOura("u1"); expect(windowDays()).toBe(7); - // One indexed read per tick, and tombstones are excluded from it — a - // soft-deleted row must not pass for a live one. - expect(findFirstMock).toHaveBeenCalledTimes(1); + // One indexed read per tick for the day window (the heart-rate window + // reads its own, typed watermark), and tombstones are excluded from it — + // a soft-deleted row must not pass for a live one. + expect(dayWatermarkReads()).toHaveLength(1); expect(findFirstMock).toHaveBeenCalledWith({ where: { userId: "u1", source: "OURA", deletedAt: null }, orderBy: { measuredAt: "desc" }, @@ -958,7 +976,7 @@ describe("syncUserOura — window resolution", () => { await syncUserOura("u1", { lookbackDays: 3 }); expect(windowDays()).toBe(3); - expect(findFirstMock).not.toHaveBeenCalled(); + expect(dayWatermarkReads()).toHaveLength(0); }); it("passes the resolved window to the cycle-phases leg too", async () => { @@ -1113,6 +1131,8 @@ describe("syncUserOura — a grant without the spo2 / heart_health scopes", () = fetchVo2MaxMock, fetchCardioAgeMock, fetchResilienceMock, + fetchWorkoutsMock, + heartRatePagesMock, ]) { m.mockRejectedValue(err401()); } @@ -1286,3 +1306,42 @@ describe("syncUserOura — Oura names the missing scope", () => { ); }); }); + +describe("syncUserOura — a grant from before workouts and heart rate were read", () => { + it("notes the two collections calmly: no refresh, no failure, everything else imports", async () => { + getConnMock.mockResolvedValue(CONN); + fetchActivityMock.mockResolvedValue([ + { id: "a1", day: "2026-06-10", steps: 1200, active_calories: 80 }, + ]); + fetchWorkoutsMock.mockRejectedValue(errScope401("workout")); + heartRatePagesMock.mockRejectedValue(errScope401("heartrate")); + + const result = await syncUserOura("u1"); + + expect(result.failed).toBe(false); + expect(result.imported).toBeGreaterThan(0); + expect(refreshMock).not.toHaveBeenCalled(); + expect(recordFailureMock).not.toHaveBeenCalled(); + expect(recordSuccessMock).toHaveBeenCalledWith("u1", "oura"); + expect(scopeDeniedMock).toHaveBeenCalledWith( + "u1", + expect.arrayContaining(["workout", "heartrate"]), + ); + expect(scopeDeniedMock.mock.calls[0]![1]).toHaveLength(2); + }); + + it("reads heart rate over a window of hours, not the day window", async () => { + getConnMock.mockResolvedValue(CONN); + findFirstMock.mockResolvedValue(null); + + await syncUserOura("u1"); + + const range = heartRatePagesMock.mock.calls[0]![1] as { + start: Date; + end: Date; + }; + expect(range.end.getTime() - range.start.getTime()).toBeLessThanOrEqual( + 72 * 60 * 60 * 1000, + ); + }); +}); diff --git a/src/lib/oura/__tests__/workouts-heart-rate.test.ts b/src/lib/oura/__tests__/workouts-heart-rate.test.ts new file mode 100644 index 000000000..7460e99f1 --- /dev/null +++ b/src/lib/oura/__tests__/workouts-heart-rate.test.ts @@ -0,0 +1,298 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { createManyAndReturnMock, findFirstMock, updateMock, safeFetchMock } = + vi.hoisted(() => ({ + createManyAndReturnMock: vi.fn(), + findFirstMock: vi.fn(), + updateMock: vi.fn(), + safeFetchMock: vi.fn(), + })); + +vi.mock("@/lib/db", () => ({ + prisma: { + measurement: { + createManyAndReturn: createManyAndReturnMock, + findFirst: findFirstMock, + }, + workout: { + createManyAndReturn: createManyAndReturnMock, + update: updateMock, + }, + }, +})); +vi.mock("@/lib/safe-fetch", () => ({ safeFetch: safeFetchMock })); +vi.mock("@/lib/arrivals/measurement-emit", () => ({ + emitInsertedMeasurementArrivals: vi.fn(async () => {}), +})); +vi.mock("@/lib/arrivals/workout-emit", () => ({ + emitInsertedWorkoutArrival: vi.fn(async () => {}), +})); +vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({ + ...(await importOriginal< + typeof import("@/lib/rollups/measurement-rollups") + >()), + recomputeBucketsForKeys: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); + +import { mapHeartRateSample } from "../client"; +import { OuraApiError } from "../response-classifier"; +import { mapOuraSportType } from "../sport-map"; +import { + OURA_HEART_RATE_LOOKBACK_HOURS, + OURA_HEART_RATE_MAX_LOOKBACK_HOURS, + OURA_HEART_RATE_WRITE_CHUNK, + resolveOuraHeartRateWindow, + syncOuraHeartRateRange, +} from "../sync-heart-rate"; +import { mapOuraWorkout, upsertOuraWorkouts } from "../sync-workouts"; + +const HOUR = 60 * 60 * 1000; + +function jsonResponse(body: unknown, status = 200, headers = {}) { + return new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json", ...headers }, + }); +} + +beforeEach(() => { + createManyAndReturnMock.mockReset(); + findFirstMock.mockReset(); + updateMock.mockReset(); + safeFetchMock.mockReset(); +}); + +describe("mapOuraSportType", () => { + it("maps Oura's labels onto the canonical buckets", () => { + expect(mapOuraSportType("walking")).toBe("walking"); + expect(mapOuraSportType("strength_training")).toBe("strength"); + expect(mapOuraSportType("Strength Training")).toBe("strength"); + expect(mapOuraSportType("cycling")).toBe("cycling"); + expect(mapOuraSportType("kayaking")).toBe("rowing"); + expect(mapOuraSportType("cross_country_skiing")).toBe("other"); + }); + + it("lands an unknown or blank label on other", () => { + expect(mapOuraSportType("underwater_hockey")).toBe("other"); + expect(mapOuraSportType("")).toBe("other"); + expect(mapOuraSportType(null)).toBe("other"); + }); +}); + +describe("mapOuraWorkout", () => { + const base = { + id: "w-1", + activity: "cycling", + calories: 412.6, + distance: 18250.4, + intensity: "moderate", + source: "autodetected", + label: "Commute with a friend", + start_datetime: "2026-10-01T07:00:00+02:00", + end_datetime: "2026-10-01T07:45:30+02:00", + }; + + it("maps times, energy, real distance and provenance", () => { + expect(mapOuraWorkout(base)).toEqual({ + externalId: "w-1", + sportType: "cycling", + startedAt: new Date("2026-10-01T05:00:00Z"), + endedAt: new Date("2026-10-01T05:45:30Z"), + durationSec: 2730, + totalEnergyKcal: 413, + totalDistanceM: 18250.4, + metadata: { + ouraActivity: "cycling", + ouraIntensity: "moderate", + ouraSource: "autodetected", + }, + }); + }); + + it("never carries the person's free-text label", () => { + expect(JSON.stringify(mapOuraWorkout(base))).not.toContain("Commute"); + }); + + it("refuses a record without an id or with an end before its start", () => { + expect(mapOuraWorkout({ ...base, id: "" })).toBeNull(); + expect( + mapOuraWorkout({ ...base, end_datetime: "2026-10-01T06:00:00+02:00" }), + ).toBeNull(); + expect(mapOuraWorkout({ ...base, start_datetime: "nope" })).toBeNull(); + }); + + it("keeps a missing energy or distance null rather than zero", () => { + const row = mapOuraWorkout({ ...base, calories: null, distance: -1 }); + expect(row?.totalEnergyKcal).toBeNull(); + expect(row?.totalDistanceM).toBeNull(); + }); +}); + +describe("upsertOuraWorkouts", () => { + it("inserts a new workout and overwrites an existing one in place", async () => { + const row = mapOuraWorkout({ + id: "w-1", + activity: "running", + start_datetime: "2026-10-01T07:00:00Z", + end_datetime: "2026-10-01T07:30:00Z", + })!; + createManyAndReturnMock + .mockResolvedValueOnce([{ id: "x", startedAt: row.startedAt }]) + .mockResolvedValueOnce([]); + expect(await upsertOuraWorkouts("u1", [row])).toBe(1); + expect(createManyAndReturnMock.mock.calls[0]![0].data).toMatchObject({ + userId: "u1", + source: "OURA", + externalId: "w-1", + sportType: "running", + }); + expect(updateMock).not.toHaveBeenCalled(); + + expect(await upsertOuraWorkouts("u1", [row])).toBe(1); + expect(updateMock).toHaveBeenCalledWith( + expect.objectContaining({ + where: { + userId_source_externalId: { + userId: "u1", + source: "OURA", + externalId: "w-1", + }, + }, + }), + ); + }); +}); + +describe("mapHeartRateSample", () => { + it("maps a sample to a PULSE row keyed on its instant", () => { + expect( + mapHeartRateSample({ + bpm: 61.4, + source: "awake", + timestamp: "2026-10-01T10:05:00+00:00", + }), + ).toEqual({ + type: "PULSE", + value: 61, + unit: "bpm", + measuredAt: new Date("2026-10-01T10:05:00Z"), + fieldTag: "hr", + externalId: "heartrate:2026-10-01T10:05:00.000Z", + }); + }); + + it("drops a sample without an instant or outside a human band", () => { + expect(mapHeartRateSample({ bpm: 60, timestamp: null })).toBeNull(); + expect(mapHeartRateSample({ bpm: 60, timestamp: "x" })).toBeNull(); + expect( + mapHeartRateSample({ bpm: 0, timestamp: "2026-10-01T10:05:00Z" }), + ).toBeNull(); + expect( + mapHeartRateSample({ bpm: 400, timestamp: "2026-10-01T10:05:00Z" }), + ).toBeNull(); + }); +}); + +describe("syncOuraHeartRateRange", () => { + it("writes every page as it arrives, in bounded batches, following next_token", async () => { + const start = new Date("2026-10-01T00:00:00Z"); + const page = (offset: number, n: number) => + Array.from({ length: n }, (_, i) => ({ + bpm: 60, + source: "awake", + timestamp: new Date( + start.getTime() + (offset + i) * 60_000, + ).toISOString(), + })); + safeFetchMock + .mockResolvedValueOnce( + jsonResponse({ data: page(0, 1200), next_token: "p2" }), + ) + .mockResolvedValueOnce( + jsonResponse({ data: page(1200, 10), next_token: null }), + ); + createManyAndReturnMock.mockImplementation( + async ({ data }: { data: Array<{ measuredAt: Date }> }) => + data.map((d, i) => ({ id: `r${i}`, type: "PULSE", ...d })), + ); + + const result = await syncOuraHeartRateRange("u1", "tok", { + start, + end: new Date(start.getTime() + 2 * 24 * HOUR), + }); + + expect(result).toEqual({ imported: 1210, truncated: false }); + // 1 200 samples in chunks of 500, then the second page. + expect( + createManyAndReturnMock.mock.calls.map(([a]) => a.data.length), + ).toEqual([ + OURA_HEART_RATE_WRITE_CHUNK, + OURA_HEART_RATE_WRITE_CHUNK, + 200, + 10, + ]); + expect(createManyAndReturnMock.mock.calls[0]![0].skipDuplicates).toBe(true); + const firstUrl = new URL(safeFetchMock.mock.calls[0]![0] as string); + expect(firstUrl.pathname).toBe("/v2/usercollection/heartrate"); + expect(firstUrl.searchParams.get("start_datetime")).toBe( + start.toISOString(), + ); + const secondUrl = new URL(safeFetchMock.mock.calls[1]![0] as string); + expect(secondUrl.searchParams.get("next_token")).toBe("p2"); + }); + + it("carries Retry-After on a 429", async () => { + safeFetchMock.mockResolvedValueOnce( + jsonResponse({ detail: "slow down" }, 429, { "retry-after": "120" }), + ); + const err = await syncOuraHeartRateRange("u1", "tok", { + start: new Date("2026-10-01T00:00:00Z"), + end: new Date("2026-10-02T00:00:00Z"), + }).catch((e: unknown) => e); + expect(err).toBeInstanceOf(OuraApiError); + expect((err as OuraApiError).retryAfterMs).toBe(120_000); + expect((err as OuraApiError).classification).toBe("transient"); + }); +}); + +describe("resolveOuraHeartRateWindow", () => { + const now = new Date("2026-10-21T12:00:00Z"); + + it("reads the last few hours while readings are flowing", async () => { + findFirstMock.mockResolvedValue({ + measuredAt: new Date(now.getTime() - 20 * 60_000), + }); + const w = await resolveOuraHeartRateWindow("u1", now); + expect(now.getTime() - w.start.getTime()).toBe( + OURA_HEART_RATE_LOOKBACK_HOURS * HOUR, + ); + expect(w.end).toEqual(now); + }); + + it("reaches back to the newest reading after a gap, up to the ceiling", async () => { + findFirstMock.mockResolvedValueOnce({ + measuredAt: new Date(now.getTime() - 30 * HOUR), + }); + let w = await resolveOuraHeartRateWindow("u1", now); + expect(now.getTime() - w.start.getTime()).toBe(31 * HOUR); + + findFirstMock.mockResolvedValueOnce({ + measuredAt: new Date(now.getTime() - 30 * 24 * HOUR), + }); + w = await resolveOuraHeartRateWindow("u1", now); + expect(now.getTime() - w.start.getTime()).toBe( + OURA_HEART_RATE_MAX_LOOKBACK_HOURS * HOUR, + ); + }); + + it("leaves history to the backfill on a fresh connection", async () => { + findFirstMock.mockResolvedValue(null); + const w = await resolveOuraHeartRateWindow("u1", now); + expect(now.getTime() - w.start.getTime()).toBe( + OURA_HEART_RATE_LOOKBACK_HOURS * HOUR, + ); + }); +}); 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/client.ts b/src/lib/oura/client.ts index ed99a0505..99e7140c2 100644 --- a/src/lib/oura/client.ts +++ b/src/lib/oura/client.ts @@ -23,6 +23,7 @@ import { OuraApiError, classifyOuraResponse } from "./response-classifier"; import { envOr, envValue } from "@/lib/env"; import { importValue } from "@/lib/integrations/import-value"; import { walkPages } from "@/lib/integrations/page-walk"; +import { retryAfterMsOf } from "@/lib/integrations/oauth-kit/retry-after"; import { OURA_OAUTH_SCOPE, parseOuraMissingScope } from "./scope"; const OURA_API_BASE = "https://api.ouraring.com"; @@ -449,7 +450,11 @@ export interface OuraDailyActivity { steps?: number | null; active_calories?: number | null; total_calories?: number | null; - /** Equivalent walking distance for the day, metres. */ + /** + * Oura's activity-equivalent distance, metres: what a walk of the day's + * effort would have covered. Not distance travelled; never imported (see + * `mapDailyActivity`). + */ equivalent_walking_distance?: number | null; } @@ -459,6 +464,13 @@ interface DateRangeQuery { maxPages?: number; } +/** A time-range read (`heartrate`): instants, not calendar dates. */ +export interface DateTimeRangeQuery { + start: Date; + end: Date; + maxPages?: number; +} + /** * The error for a refused collection read. Oura's status alone does not say * what went wrong; its body does (see `./scope`): @@ -469,13 +481,16 @@ interface DateRangeQuery { * - any other 401: the token is dead or expired (refresh, then reconnect). * - 403: a lapsed Oura membership, per the API reference. A reconnect does * not fix that, so it is a persistent failure rather than a reauth. + * - 429 (and 503): transient, with the `Retry-After` delay on + * `retryAfterMs` so a backfill can wait it out instead of guessing. */ function collectionRefusal( verb: string, - status: number, + res: Pick, verdict: ReturnType, body: unknown, ): OuraApiError { + const status = res.status; if (status === 401) { const missingScope = parseOuraMissingScope(body); if (missingScope) { @@ -501,9 +516,43 @@ function collectionRefusal( classification: verdict.classification, httpStatus: verdict.httpStatus, reason: verdict.reason, + ...(status === 429 || status === 503 + ? { retryAfterMs: retryAfterMsOf(res) } + : {}), }); } +/** Read one page of a collection; throws the classified refusal. */ +async function readCollectionPage( + path: string, + accessToken: string, + verb: string, + params: URLSearchParams, + page: number, +): Promise<{ items: T[]; next: string | null }> { + const pageStart = performance.now(); + const res = await safeFetch(`${OURA_API_BASE}${path}?${params}`, { + method: "GET", + headers: { + Authorization: `Bearer ${accessToken}`, + Accept: "application/json", + }, + }); + const json = (await res.json().catch(() => null)) as OuraCollection | null; + const verdict = classifyOuraResponse(res.status); + getEvent()?.addExternalCall({ + service: "oura", + method: `${verb}(page=${page})`, + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw collectionRefusal(verb, res, verdict, json); + } + return { items: json?.data ?? [], next: json?.next_token ?? null }; +} + async function fetchCollection( path: string, accessToken: string, @@ -513,42 +562,61 @@ async function fetchCollection( const walk = await walkPages( `oura.${verb}`, query.maxPages ?? 100, - async (nextToken, page) => { + (nextToken, page) => { const params = new URLSearchParams({ start_date: query.startDate, end_date: query.endDate, }); if (nextToken) params.set("next_token", nextToken); - - const pageStart = performance.now(); - const res = await safeFetch(`${OURA_API_BASE}${path}?${params}`, { - method: "GET", - headers: { - Authorization: `Bearer ${accessToken}`, - Accept: "application/json", - }, - }); - const json = (await res - .json() - .catch(() => null)) as OuraCollection | null; - const verdict = classifyOuraResponse(res.status); - getEvent()?.addExternalCall({ - service: "oura", - method: `${verb}(page=${page})`, - duration_ms: Math.round(performance.now() - pageStart), - status: res.status, - error: - verdict.classification === "success" ? undefined : verdict.reason, - }); - if (verdict.classification !== "success") { - throw collectionRefusal(verb, res.status, verdict, json); - } - return { items: json?.data ?? [], next: json?.next_token ?? null }; + return readCollectionPage(path, accessToken, verb, params, page); }, ); return walk.items; } +/** What a streamed walk came to. */ +export interface OuraStreamedWalk { + pages: number; + /** Stopped at the page ceiling or on a repeated cursor with pages left. */ + truncated: boolean; +} + +/** + * Walk a time-range collection page by page, handing each page to `onPage` + * before the next is read. Nothing is collected: the walk keeps one page at a + * time, so a dense stream (heart rate, a sample every few minutes for years) + * costs the memory of a page, not of the history. + */ +async function streamCollection( + path: string, + accessToken: string, + verb: string, + query: DateTimeRangeQuery, + onPage: (items: T[]) => Promise, +): Promise { + const walk = await walkPages( + `oura.${verb}`, + query.maxPages ?? 200, + async (nextToken, page) => { + const params = new URLSearchParams({ + start_datetime: query.start.toISOString(), + end_datetime: query.end.toISOString(), + }); + if (nextToken) params.set("next_token", nextToken); + const { items, next } = await readCollectionPage( + path, + accessToken, + verb, + params, + page, + ); + await onPage(items); + return { items: [], next }; + }, + ); + return { pages: walk.pages, truncated: walk.truncated }; +} + export function fetchSleep( accessToken: string, query: DateRangeQuery, @@ -665,6 +733,68 @@ export function fetchDailyCyclePhases( ); } +/** + * Oura `workout` — one activity session, auto-detected or entered. `activity` + * is Oura's own lowercase label (`walking`, `cycling`, `strength_training`, + * ...); `distance` is metres actually covered (a GPS or step-derived figure, + * unlike the daily activity-equivalent distance); `calories` is kcal. + */ +export interface OuraWorkout { + id: string; + day?: string; + activity?: string | null; + calories?: number | null; + distance?: number | null; + start_datetime: string; + end_datetime: string; + /** `easy` | `moderate` | `hard` */ + intensity?: string | null; + /** `manual` | `autodetected` | `confirmed` | `workout_heart_rate` */ + source?: string | null; + /** A name the person gave the workout. Free text; never imported. */ + label?: string | null; +} + +export function fetchWorkouts( + accessToken: string, + query: DateRangeQuery, +): Promise { + return fetchCollection( + "/v2/usercollection/workout", + accessToken, + "fetchWorkouts", + query, + ); +} + +/** One `heartrate` sample: a reading at an instant, tagged with its context. */ +export interface OuraHeartRateSample { + bpm?: number | null; + /** `awake` | `rest` | `sleep` | `session` | `live` | `workout` */ + source?: string | null; + timestamp?: string | null; +} + +/** + * Stream the `heartrate` collection for a time range, one page at a time. + * Oura answers it by instant (`start_datetime` / `end_datetime`) and pages + * with `next_token`; a page goes to `onPage` and is dropped before the next + * is read. + */ +export function forEachHeartRatePage( + accessToken: string, + query: DateTimeRangeQuery, + onPage: (samples: OuraHeartRateSample[]) => Promise, +): Promise { + return streamCollection( + "/v2/usercollection/heartrate", + accessToken, + "fetchHeartRate", + query, + onPage, + ); +} + // ─── Field → Measurement mapping ─────────────────────────────── const SEC_TO_MIN = 1 / 60; @@ -972,8 +1102,17 @@ export function mapDailySpo2( ]; } -/** Map one Oura daily-activity record: steps, active energy, and the - * equivalent walking distance (B5 — the OURA distance ladder slot was dead). */ +/** + * Map one Oura daily-activity record: steps and active energy. + * + * `equivalent_walking_distance` is deliberately not mapped. Oura derives it + * from the day's whole activity (the distance a walk of the same effort would + * have covered), so an hour of cycling or yoga adds kilometres nobody walked. + * Stored as `WALKING_RUNNING_DISTANCE` it read as real distance, and on a day + * no other source covered it became the day's distance. The rows earlier + * syncs wrote under `activity::distance` are retired on their own + * (`./retire-distance`). + */ export function mapDailyActivity( a: OuraDailyActivity, tz?: string, @@ -999,20 +1138,6 @@ export function mapDailyActivity( fieldTag: "active_energy", }); } - if ( - typeof a.equivalent_walking_distance === "number" && - a.equivalent_walking_distance >= 0 - ) { - out.push({ - type: "WALKING_RUNNING_DISTANCE", - // Oura reports the equivalent walking distance in metres — the canonical - // WALKING_RUNNING_DISTANCE unit (no conversion). - value: importValue(a.equivalent_walking_distance), - unit: "m", - measuredAt, - fieldTag: "distance", - }); - } return out; } @@ -1106,6 +1231,38 @@ export function mapResilience( ]; } +/** Heart-rate readings outside this band are not a ring's reading. */ +const OURA_BPM_MIN = 20; +const OURA_BPM_MAX = 250; + +/** + * Map one `heartrate` sample → `PULSE` (bpm), the intraday heart-rate type + * Google Health and Fitbit write their minute stream to. Keyed on the + * sample's own instant (`heartrate:`): Oura emits one reading per + * instant, and a sample is final once it is in the cloud, so a re-read lands + * on the same key. A sample without a parseable instant or with a bpm + * outside a human band maps to nothing. + */ +export function mapHeartRateSample( + sample: OuraHeartRateSample, +): MappedMeasurement | null { + const bpm = sample.bpm; + if (typeof bpm !== "number" || !Number.isFinite(bpm)) return null; + if (bpm < OURA_BPM_MIN || bpm > OURA_BPM_MAX) return null; + if (typeof sample.timestamp !== "string") return null; + const at = new Date(sample.timestamp); + if (Number.isNaN(at.getTime())) return null; + const iso = at.toISOString(); + return { + type: "PULSE", + value: Math.round(bpm), + unit: "bpm", + measuredAt: at, + fieldTag: "hr", + externalId: `heartrate:${iso}`, + }; +} + // ─── Cycle-phase → CycleDayLog mapping (v1.29.x) ─────────────────── // // `daily_cycle_phases` does NOT feed `Measurement` like every collection diff --git a/src/lib/oura/response-classifier.ts b/src/lib/oura/response-classifier.ts index 46a748ba0..f3c378e56 100644 --- a/src/lib/oura/response-classifier.ts +++ b/src/lib/oura/response-classifier.ts @@ -36,6 +36,12 @@ export class OuraApiError extends IntegrationApiError { * scope."}`). Undefined for every other failure, a dead token included. */ readonly missingScope: string | undefined; + /** + * How long Oura asked to be left alone, from the `Retry-After` of a 429 (or + * 503). Undefined when the refusal carried none or was not a rate limit. + * A backfill waits at least this long before its next slice. + */ + readonly retryAfterMs: number | undefined; constructor(opts: { verb: string; @@ -44,8 +50,9 @@ export class OuraApiError extends IntegrationApiError { reason: string; upstreamError?: string; missingScope?: string; + retryAfterMs?: number; }) { - const { missingScope, ...rest } = opts; + const { missingScope, retryAfterMs, ...rest } = opts; super({ vendor: "Oura", ...rest, @@ -55,6 +62,7 @@ export class OuraApiError extends IntegrationApiError { }); this.name = "OuraApiError"; this.missingScope = missingScope; + this.retryAfterMs = retryAfterMs; } } 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/scope.ts b/src/lib/oura/scope.ts index 8356c6246..7189c22ff 100644 --- a/src/lib/oura/scope.ts +++ b/src/lib/oura/scope.ts @@ -34,6 +34,8 @@ * vo2max vO2_max heart_health * cardio_age daily_cardiovascular_age heart_health * resilience daily_resilience stress + * workout workout workout + * heartrate heartrate heartrate * (profile) personal_info personal * (cycle) daily_cycle_phases reproductive_cycle * @@ -53,8 +55,12 @@ export const OURA_COLLECTIONS = [ { name: "vo2max", scope: "heart_health" }, { name: "cardio_age", scope: "heart_health" }, { name: "resilience", scope: "stress" }, + { name: "workout", scope: "workout" }, + { name: "heartrate", scope: "heartrate" }, ] as const; +export type OuraCollectionName = (typeof OURA_COLLECTIONS)[number]["name"]; + type OuraCollectionEntry = (typeof OURA_COLLECTIONS)[number]; /** The scope the profile probe (`personal_info`) needs. */ diff --git a/src/lib/oura/sport-map.ts b/src/lib/oura/sport-map.ts new file mode 100644 index 000000000..bed526098 --- /dev/null +++ b/src/lib/oura/sport-map.ts @@ -0,0 +1,133 @@ +/** + * Oura workout `activity` → HealthLog canonical `WorkoutSportType`. + * + * Oura labels a workout with its own lowercase snake_case vocabulary + * (`walking`, `cycling`, `strength_training`, `cross_country_skiing`, ...) + * and publishes no closed list of it. The table below covers the activities + * the Oura app offers; the bucket choices follow the Polar and Strava tables + * (paddle sports → rowing, racket sports → tennis, climbing and functional + * training → crossTraining, mobility work → mindAndBody, snow sports → other), + * so the same real-world session lands in the same bucket whichever source + * reported it. Anything unknown lands on `"other"`, the row still persists, + * and the raw label is kept in `Workout.metadata.ouraActivity`. + */ +import type { WorkoutSportType } from "@/lib/validations/workout"; + +const OURA_SPORT_TABLE: ReadonlyArray = [ + // running + ["running", "running"], + ["jogging", "running"], + ["trail_running", "running"], + ["treadmill_running", "running"], + // cycling + ["cycling", "cycling"], + ["indoor_cycling", "cycling"], + ["mountain_biking", "cycling"], + ["road_cycling", "cycling"], + ["e_biking", "cycling"], + ["spinning", "cycling"], + // walking / hiking + ["walking", "walking"], + ["nordic_walking", "walking"], + ["treadmill_walking", "walking"], + ["hiking", "hiking"], + ["backpacking", "hiking"], + // water + ["swimming", "swimming"], + ["open_water_swimming", "swimming"], + ["rowing", "rowing"], + ["indoor_rowing", "rowing"], + ["kayaking", "rowing"], + ["canoeing", "rowing"], + ["paddling", "rowing"], + ["stand_up_paddling", "rowing"], + // machines + ["elliptical", "elliptical"], + ["cross_trainer", "elliptical"], + ["stair_exercise", "stairClimber"], + ["stair_climbing", "stairClimber"], + ["stairs", "stairClimber"], + // strength / intervals / functional + ["strength_training", "strength"], + ["weight_training", "strength"], + ["weightlifting", "strength"], + ["core_training", "strength"], + ["hiit", "hiit"], + ["high_intensity_interval_training", "hiit"], + ["circuit_training", "hiit"], + ["cross_training", "crossTraining"], + ["crossfit", "crossTraining"], + ["functional_training", "crossTraining"], + ["climbing", "crossTraining"], + ["bouldering", "crossTraining"], + ["rock_climbing", "crossTraining"], + ["mixed_cardio", "mixedCardio"], + ["cardio", "mixedCardio"], + ["aerobics", "dance"], + // mind and body + ["yoga", "yoga"], + ["hot_yoga", "yoga"], + ["pilates", "mindAndBody"], + ["stretching", "mindAndBody"], + ["mobility", "mindAndBody"], + ["barre", "mindAndBody"], + ["tai_chi", "mindAndBody"], + // dance + ["dancing", "dance"], + ["dance", "dance"], + ["zumba", "dance"], + // ball and racket sports + ["golf", "golf"], + ["tennis", "tennis"], + ["table_tennis", "tennis"], + ["squash", "tennis"], + ["padel", "tennis"], + ["pickleball", "tennis"], + ["racquetball", "tennis"], + ["badminton", "badminton"], + ["basketball", "basketball"], + ["soccer", "soccer"], + ["football", "soccer"], + // everything else that persists outside a bucket + ["skiing", "other"], + ["cross_country_skiing", "other"], + ["downhill_skiing", "other"], + ["snowboarding", "other"], + ["skating", "other"], + ["ice_skating", "other"], + ["surfing", "other"], + ["boxing", "other"], + ["martial_arts", "other"], + ["volleyball", "other"], + ["hockey", "other"], + ["housework", "other"], + ["gardening", "other"], + ["other", "other"], +]; + +/** Fold case and drop separators, so `Strength Training` meets `strength_training`. */ +function normaliseOuraActivity(raw: string): string { + return raw + .trim() + .toLowerCase() + .replace(/[^a-z0-9]+/g, ""); +} + +const OURA_SPORT_MAP: ReadonlyMap = new Map( + OURA_SPORT_TABLE.map(([name, canonical]) => [ + normaliseOuraActivity(name), + canonical, + ]), +); + +/** + * Resolve an Oura workout's `activity` to a canonical `WorkoutSportType`; + * `"other"` for an absent, blank or unknown label. Never returns the raw + * label. + */ +export function mapOuraSportType( + activity: string | null | undefined, +): WorkoutSportType { + if (typeof activity !== "string" || activity.trim() === "") return "other"; + return OURA_SPORT_MAP.get(normaliseOuraActivity(activity)) ?? "other"; +} diff --git a/src/lib/oura/sync-heart-rate.ts b/src/lib/oura/sync-heart-rate.ts new file mode 100644 index 000000000..17597a23e --- /dev/null +++ b/src/lib/oura/sync-heart-rate.ts @@ -0,0 +1,200 @@ +/** + * Oura intraday heart rate (`/v2/usercollection/heartrate`) → `PULSE`. + * + * The ring writes a reading every few minutes through the day, more during a + * workout or a session, so a history of years is hundreds of thousands of + * rows. The walk is streamed (`forEachHeartRatePage`): each page is mapped and + * written in bounded batches before the next page is read, and nothing of a + * page outlives its write. What the window keeps is the set of touched days + * (a handful per window), which the caller refolds once at the end. + * + * Writes are first-write-wins: a sample is final once it is in Oura's cloud, + * and it is keyed on its own instant (`heartrate:`), so a re-read of the + * same range is an insert that `skipDuplicates` drops. A row the person + * deleted keeps its tombstone; the re-read does not bring it back. Oura's + * samples go raw, like Google Health's and Fitbit's minute stream: the dense + * hourly fold covers only the export sources (`DENSE_INTRADAY_FOLD_SOURCES`). + */ +import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; +import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; +import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; +import { + collapseToTypeDayKeys, + recomputeBucketsForKeys, +} from "@/lib/rollups/measurement-rollups"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { getEvent } from "@/lib/logging/context"; +import { + forEachHeartRatePage, + mapHeartRateSample, + type MappedMeasurement, + type OuraHeartRateSample, +} from "./client"; + +/** Rows per `createMany`: bounded whatever size of page Oura sends. */ +export const OURA_HEART_RATE_WRITE_CHUNK = 500; + +/** + * The hourly window: how far back an hourly sync reads heart rate. Short on + * purpose, the history belongs to the backfill. A ring that has not synced + * for a while widens it up to the ceiling, anchored on the newest reading + * already held (`resolveOuraHeartRateWindow`). + */ +export const OURA_HEART_RATE_LOOKBACK_HOURS = 6; +export const OURA_HEART_RATE_MAX_LOOKBACK_HOURS = 72; + +const HOUR_MS = 60 * 60 * 1000; + +/** Day keys touched by a write, collapsed as they arrive. */ +export class TouchedDays { + private readonly keys = new Map< + string, + { type: MeasurementType; measuredAt: Date } + >(); + + add(rows: Array<{ type: MeasurementType; measuredAt: Date }>): void { + for (const k of collapseToTypeDayKeys(rows)) { + this.keys.set(`${k.type}|${k.measuredAt.getTime()}`, k); + } + } + + get size(): number { + return this.keys.size; + } + + list(): Array<{ type: MeasurementType; measuredAt: Date }> { + return [...this.keys.values()]; + } +} + +/** + * Write one page of samples. Returns how many rows were inserted; the days + * the page reached go into `touched`. + */ +export async function writeOuraHeartRatePage( + userId: string, + samples: readonly OuraHeartRateSample[], + touched: TouchedDays, + opts: { emitArrivals?: boolean } = {}, +): Promise { + const mapped: MappedMeasurement[] = []; + for (const s of samples) { + const m = mapHeartRateSample(s); + if (m) mapped.push(m); + } + const kept = dropImplausibleMeasurements("oura", mapped, (m) => ({ + type: m.type, + value: m.value, + })); + let inserted = 0; + for (let i = 0; i < kept.length; i += OURA_HEART_RATE_WRITE_CHUNK) { + const chunk = kept.slice(i, i + OURA_HEART_RATE_WRITE_CHUNK); + const rows = await prisma.measurement.createManyAndReturn({ + data: chunk.map((m) => ({ + userId, + type: m.type as MeasurementType, + source: "OURA" as const, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.externalId!, + })), + skipDuplicates: true, + select: { id: true, type: true, measuredAt: true }, + }); + inserted += rows.length; + touched.add(rows); + if (opts.emitArrivals !== false && rows.length > 0) { + void emitInsertedMeasurementArrivals(userId, rows, "oura").catch( + () => {}, + ); + } + } + return inserted; +} + +/** Refold the touched days once a window is written. Best-effort. */ +export async function refoldOuraTouchedDays( + userId: string, + touched: TouchedDays, +): Promise { + if (touched.size === 0) return; + const keys = touched.list(); + try { + await recomputeBucketsForKeys(userId, keys); + invalidateStatusInsightsForTypes( + userId, + [...new Set(keys.map((k) => k.type))], + { source: "sync" }, + ).catch((err) => { + getEvent()?.addWarning( + `oura: status-insight invalidate failed for ${userId}: ${err}`, + ); + }); + } catch (err) { + getEvent()?.addWarning( + `oura: heart-rate rollup recompute failed for ${userId}: ${err}`, + ); + } +} + +/** + * Read and write one time range of heart rate, page by page. Returns how many + * rows were inserted and whether the walk stopped short (page ceiling or a + * cursor Oura handed out twice). A refused page throws the classified + * `OuraApiError` (a 429 with its `retryAfterMs`); the pages written before it + * stay written. + */ +export async function syncOuraHeartRateRange( + userId: string, + accessToken: string, + range: { start: Date; end: Date }, + opts: { emitArrivals?: boolean; touched?: TouchedDays } = {}, +): Promise<{ imported: number; truncated: boolean }> { + const touched = opts.touched ?? new TouchedDays(); + let imported = 0; + const walk = await forEachHeartRatePage( + accessToken, + { start: range.start, end: range.end }, + async (samples) => { + imported += await writeOuraHeartRatePage(userId, samples, touched, { + emitArrivals: opts.emitArrivals, + }); + }, + ); + if (!opts.touched) await refoldOuraTouchedDays(userId, touched); + return { imported, truncated: walk.truncated }; +} + +/** + * The hourly heart-rate window: the last `OURA_HEART_RATE_LOOKBACK_HOURS`, or + * back to an hour before the newest reading held when that is older, never + * further than `OURA_HEART_RATE_MAX_LOOKBACK_HOURS`. The ring uploads what it + * stored all at once when the app next opens, so a gap closes in one window. + */ +export async function resolveOuraHeartRateWindow( + userId: string, + now: Date = new Date(), +): Promise<{ start: Date; end: Date }> { + const newest = await prisma.measurement.findFirst({ + where: { + userId, + source: "OURA", + type: "PULSE", + deletedAt: null, + externalId: { startsWith: "heartrate:" }, + }, + orderBy: { measuredAt: "desc" }, + select: { measuredAt: true }, + }); + const floor = now.getTime() - OURA_HEART_RATE_MAX_LOOKBACK_HOURS * HOUR_MS; + const steady = now.getTime() - OURA_HEART_RATE_LOOKBACK_HOURS * HOUR_MS; + const fromNewest = newest + ? newest.measuredAt.getTime() - HOUR_MS + : Number.POSITIVE_INFINITY; + return { + start: new Date(Math.max(floor, Math.min(steady, fromNewest))), + end: now, + }; +} diff --git a/src/lib/oura/sync-workouts.ts b/src/lib/oura/sync-workouts.ts new file mode 100644 index 000000000..0d3ff8def --- /dev/null +++ b/src/lib/oura/sync-workouts.ts @@ -0,0 +1,150 @@ +/** + * Oura workouts → `Workout` rows (`source = OURA`). + * + * Each Oura workout becomes one row keyed on `(userId, source, externalId)` + * with the Oura id as `externalId`, so a re-read (the hourly window, a + * backfill slice, a workout Oura revised) overwrites in place. The write is + * the same `createManyAndReturn` + `skipDuplicates`, update-on-conflict shape + * the Polar, WHOOP and Strava ingests use; the arrival event fires only for a + * fresh insert. + * + * Cross-source dedup is not done here: an Oura-detected walk and the same walk + * from Apple Health stay two rows, and the read-time `pickCanonicalWorkoutRows` + * collapses the twin by the user's source ladder, like every other provider. + * + * `sportType` is the canonical bucket from `mapOuraSportType`; the raw label + * travels in `metadata.ouraActivity`, with Oura's intensity grade and how the + * workout was recorded (`manual`, `autodetected`, ...). The person's own name + * for the workout (`label`) is free text and is not imported. + */ +import { prisma } from "@/lib/db"; +import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit"; +import { annotate, getEvent } from "@/lib/logging/context"; +import type { Prisma } from "@/generated/prisma/client"; +import type { WorkoutSportType } from "@/lib/validations/workout"; +import type { OuraWorkout } from "./client"; +import { mapOuraSportType } from "./sport-map"; + +export interface OuraWorkoutRow { + externalId: string; + sportType: WorkoutSportType; + startedAt: Date; + endedAt: Date; + durationSec: number; + totalEnergyKcal: number | null; + totalDistanceM: number | null; + metadata: Prisma.InputJsonValue; +} + +function finiteNonNegative(v: number | null | undefined): number | null { + return typeof v === "number" && Number.isFinite(v) && v >= 0 ? v : null; +} + +/** + * Map one Oura workout to a row, or null when it has no id or no usable time + * range (an end before its start included). + */ +export function mapOuraWorkout(w: OuraWorkout): OuraWorkoutRow | null { + if (typeof w.id !== "string" || w.id === "") return null; + const startedAt = new Date(w.start_datetime); + const endedAt = new Date(w.end_datetime); + if (Number.isNaN(startedAt.getTime()) || Number.isNaN(endedAt.getTime())) { + return null; + } + if (endedAt.getTime() < startedAt.getTime()) return null; + const activity = typeof w.activity === "string" ? w.activity : null; + const metadata: Prisma.InputJsonValue = { + ...(activity ? { ouraActivity: activity } : {}), + ...(typeof w.intensity === "string" ? { ouraIntensity: w.intensity } : {}), + ...(typeof w.source === "string" ? { ouraSource: w.source } : {}), + }; + const kcal = finiteNonNegative(w.calories); + return { + externalId: w.id, + sportType: mapOuraSportType(activity), + startedAt, + endedAt, + durationSec: Math.round((endedAt.getTime() - startedAt.getTime()) / 1000), + totalEnergyKcal: kcal === null ? null : Math.round(kcal), + totalDistanceM: finiteNonNegative(w.distance), + metadata, + }; +} + +/** + * Upsert mapped Oura workouts. Every row is attempted; the first failure is + * rethrown afterwards so a partially written batch is never reported clean. + * `emitArrivals: false` (the history backfill) writes without arrival events: + * a workout from three years ago is not news. + */ +export async function upsertOuraWorkouts( + userId: string, + rows: readonly OuraWorkoutRow[], + opts: { emitArrivals?: boolean } = {}, +): Promise { + if (rows.length === 0) return 0; + const emitArrivals = opts.emitArrivals ?? true; + let imported = 0; + let firstError: unknown; + let hadError = false; + for (const r of rows) { + const data = { + sportType: r.sportType, + startedAt: r.startedAt, + endedAt: r.endedAt, + durationSec: r.durationSec, + totalEnergyKcal: r.totalEnergyKcal, + totalDistanceM: r.totalDistanceM, + metadata: r.metadata, + }; + try { + const [inserted] = await prisma.workout.createManyAndReturn({ + data: { userId, source: "OURA", externalId: r.externalId, ...data }, + skipDuplicates: true, + select: { id: true, startedAt: true }, + }); + if (inserted) { + if (emitArrivals) { + void emitInsertedWorkoutArrival(userId, inserted, "oura").catch( + () => {}, + ); + } + } else { + await prisma.workout.update({ + where: { + userId_source_externalId: { + userId, + source: "OURA", + externalId: r.externalId, + }, + }, + data, + }); + } + imported += 1; + } catch (err) { + getEvent()?.addWarning(`oura: failed to upsert workout: ${err}`); + if (!hadError) { + firstError = err; + hadError = true; + } + } + } + if (hadError) throw firstError; + annotate({ action: { name: "oura.workout.ingest", details: { imported } } }); + return imported; +} + +/** Map and write one window's workouts; unmappable records are skipped. */ +export async function writeOuraWorkouts( + userId: string, + workouts: readonly OuraWorkout[], + opts: { emitArrivals?: boolean } = {}, +): Promise { + const rows: OuraWorkoutRow[] = []; + for (const w of workouts) { + const row = mapOuraWorkout(w); + if (row) rows.push(row); + } + return upsertOuraWorkouts(userId, rows, opts); +} diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index 39aee9733..c205e05e5 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -4,7 +4,7 @@ * Pulls daily readiness (→ RECOVERY_SCORE + BODY_TEMPERATURE_DEVIATION), sleep * (real per-segment hypnogram timeline when present, else per-stage totals; * efficiency, HRV, RHR, respiratory rate), daily activity (steps, active - * energy, equivalent walking distance), the daily Sleep Score (→ SLEEP_SCORE), + * energy), the daily Sleep Score (→ SLEEP_SCORE), * daily SpO2 (→ OXYGEN_SATURATION), the dedicated vO2_max collection (→ * VO2_MAX), and the daily cardiovascular-age estimate (→ VASCULAR_AGE) for one * connected user, mapping each into `Measurement` rows tagged `source = OURA`. @@ -75,6 +75,7 @@ import { fetchResilience, fetchSleep, fetchVo2Max, + fetchWorkouts, isRetiredOuraSleepPeriod, mapCardiovascularAge, mapDailyActivity, @@ -90,7 +91,18 @@ import { getOuraConnection, syncOuraScopeDenied } from "./credentials"; import { refreshOuraAfterRefusal } from "./oauth-descriptor"; import { OuraApiError, classifyOuraError } from "./response-classifier"; import { syncUserOuraCyclePhases } from "./cycle-sync"; -import { isOuraScopeGatedCollection } from "./scope"; +import { + OURA_COLLECTIONS, + isOuraScopeGatedCollection, + type OuraCollectionName, +} from "./scope"; +import { writeOuraWorkouts } from "./sync-workouts"; +import { + refoldOuraTouchedDays, + resolveOuraHeartRateWindow, + syncOuraHeartRateRange, + TouchedDays, +} from "./sync-heart-rate"; import { shiftDateKey, userDayKey } from "@/lib/tz/format"; import { resolveUserTimezone } from "@/lib/tz/resolver"; @@ -220,9 +232,12 @@ function toUpsert( /** One Oura collection: its ledger name and a self-contained fetch+map closure * that resolves to the collection's mapped upserts. Each closure owns its own - * client call and mapper so a throw is contained to this one collection. */ + * client call and mapper so a throw is contained to this one collection. A + * collection that writes rows of its own (workouts, the heart-rate stream) + * writes inside the closure, counts on `OuraRunContext.sideImported`, and + * resolves to no upserts. */ interface OuraCollectionSpec { - name: string; + name: OuraCollectionName; collect: (token: string) => Promise; } @@ -261,12 +276,167 @@ export interface OuraFetchResult { scopeDenied: string[] | null; } +/** + * What one run reads: the day window of the day-keyed collections, the time + * range of the heart-rate stream, the user's zone, and the sinks the sleep + * collection and the self-writing collections fill as they go. + */ +export interface OuraRunContext { + userId: string; + tz: string; + query: { startDate: string; endDate: string }; + heartRate: { start: Date; end: Date }; + /** False on a history backfill: old rows are not arrivals. */ + emitArrivals: boolean; + sleepSweeps: SleepSegmentSweep[]; + sleepRetirements: string[]; + /** Rows written by the collections that write their own. */ + sideImported: number; + /** Days the heart-rate stream touched; refolded once by the caller. */ + heartRateTouched: TouchedDays; +} + +export function ouraRunContext( + userId: string, + tz: string, + query: { startDate: string; endDate: string }, + heartRate: { start: Date; end: Date }, + emitArrivals = true, +): OuraRunContext { + return { + userId, + tz, + query, + heartRate, + emitArrivals, + sleepSweeps: [], + sleepRetirements: [], + sideImported: 0, + heartRateTouched: new TouchedDays(), + }; +} + +/** + * The fetch+map (or fetch+write) closure of one collection over the context's + * window. The one place each collection's read is defined, for the hourly + * sync and the history backfill alike. + */ +export function ouraCollectionReader( + name: OuraCollectionName, + ctx: OuraRunContext, +): (token: string) => Promise { + const { tz, query } = ctx; + switch (name) { + case "readiness": + return async (t) => + (await fetchReadiness(t, query)).flatMap((r) => + toUpsert( + mapReadiness(r, tz), + "readiness", + r.timestamp ? undefined : r.day, + ), + ); + case "sleep": + return async (t) => { + const records = await fetchSleep(t, query); + const ups: OuraMeasurementUpsert[] = []; + for (const s of records) { + if (isRetiredOuraSleepPeriod(s)) { + if (s.id) ctx.sleepRetirements.push(`sleep:${s.id}:`); + continue; + } + const rows = toUpsert( + mapSleep(s, tz), + "sleep", + s.bedtime_end ? undefined : s.day, + ); + ups.push(...rows); + // Record-scoped sweep entry: every SLEEP_DURATION row of this + // record keys under `sleep::` (hypnogram runs AND the + // stage-total fallback), so the prefix bounds the cleanup to this + // one record. Clears whatever a revised hypnogram orphaned — + // shifted run boundaries, legacy run-indexed `seg:` rows, and a + // stale stage-total set once the hypnogram appears. + ctx.sleepSweeps.push({ + prefix: `sleep:${s.id}:`, + keepIds: rows + .filter((r) => r.type === "SLEEP_DURATION") + .map((r) => r.externalId), + }); + } + return ups; + }; + case "activity": + return async (t) => + (await fetchDailyActivity(t, query)).flatMap((a) => + toUpsert( + mapDailyActivity(a, tz), + "activity", + a.timestamp ? undefined : a.day, + ), + ); + case "daily_sleep": + return async (t) => + (await fetchDailySleep(t, query)).flatMap((d) => + toUpsert( + mapDailySleep(d, tz), + "daily_sleep", + d.timestamp ? undefined : d.day, + ), + ); + case "spo2": + return async (t) => + (await fetchDailySpo2(t, query)).flatMap((s) => + toUpsert(mapDailySpo2(s, tz), "spo2", s.day), + ); + case "vo2max": + return async (t) => + (await fetchVo2Max(t, query)).flatMap((v) => + toUpsert(mapVo2Max(v, tz), "vo2max", v.timestamp ? undefined : v.day), + ); + case "cardio_age": + return async (t) => + (await fetchCardiovascularAge(t, query)).flatMap((c) => + toUpsert(mapCardiovascularAge(c, tz), "cardio_age", c.day), + ); + case "resilience": + return async (t) => + (await fetchResilience(t, query)).flatMap((r) => + toUpsert(mapResilience(r, tz), "resilience", r.day), + ); + case "workout": + return async (t) => { + const workouts = await fetchWorkouts(t, query); + ctx.sideImported += await writeOuraWorkouts(ctx.userId, workouts, { + emitArrivals: ctx.emitArrivals, + }); + return []; + }; + case "heartrate": + return async (t) => { + const { imported, truncated } = await syncOuraHeartRateRange( + ctx.userId, + t, + ctx.heartRate, + { emitArrivals: ctx.emitArrivals, touched: ctx.heartRateTouched }, + ); + ctx.sideImported += imported; + if (truncated) { + // A window Oura could not page to its end is not read: fail it so + // the run is not reported clean over a gap. + throw new Error("oura heart-rate walk stopped short of its window"); + } + return []; + }; + } +} + /** * A 401 whose body names the missing scope (see `./scope`): Oura checked the * 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 && @@ -279,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 && @@ -346,124 +516,17 @@ function splitScopeDenied( * connection at `reauth_required`. */ async function fetchAll( - userId: string, + ctx: OuraRunContext, accessToken: string, refreshTokenCiphertext: string, - lookbackDays: number, knownScopeDenied: readonly string[] = [], ): Promise { - // Oura keys its daily collections by the user's calendar date, so the - // window is cut on the user's own days, and date-only records are anchored - // at local noon of their date in this zone. - const tz = await resolveUserTimezone(userId); - const today = userDayKey(new Date(), tz); - const query = { - startDate: shiftDateKey(today, -lookbackDays), - endDate: today, - }; - - // Filled by the sleep collection's closure as it maps records; carried on - // the result so the caller can run the record-scoped sweep before the - // upsert. A collection that throws pushes nothing (the fetch is its first - // await), so a 401-retry cannot double-enter a record. - const sleepSweeps: SleepSegmentSweep[] = []; - const sleepRetirements: string[] = []; - - const collections: OuraCollectionSpec[] = [ - { - name: "readiness", - collect: async (t) => - (await fetchReadiness(t, query)).flatMap((r) => - toUpsert( - mapReadiness(r, tz), - "readiness", - r.timestamp ? undefined : r.day, - ), - ), - }, - { - name: "sleep", - collect: async (t) => { - const records = await fetchSleep(t, query); - const ups: OuraMeasurementUpsert[] = []; - for (const s of records) { - if (isRetiredOuraSleepPeriod(s)) { - if (s.id) sleepRetirements.push(`sleep:${s.id}:`); - continue; - } - const rows = toUpsert( - mapSleep(s, tz), - "sleep", - s.bedtime_end ? undefined : s.day, - ); - ups.push(...rows); - // Record-scoped sweep entry: every SLEEP_DURATION row of this - // record keys under `sleep::` (hypnogram runs AND the - // stage-total fallback), so the prefix bounds the cleanup to this - // one record. Clears whatever a revised hypnogram orphaned — - // shifted run boundaries, legacy run-indexed `seg:` rows, and a - // stale stage-total set once the hypnogram appears. - sleepSweeps.push({ - prefix: `sleep:${s.id}:`, - keepIds: rows - .filter((r) => r.type === "SLEEP_DURATION") - .map((r) => r.externalId), - }); - } - return ups; - }, - }, - { - name: "activity", - collect: async (t) => - (await fetchDailyActivity(t, query)).flatMap((a) => - toUpsert( - mapDailyActivity(a, tz), - "activity", - a.timestamp ? undefined : a.day, - ), - ), - }, - { - name: "daily_sleep", - collect: async (t) => - (await fetchDailySleep(t, query)).flatMap((d) => - toUpsert( - mapDailySleep(d, tz), - "daily_sleep", - d.timestamp ? undefined : d.day, - ), - ), - }, - { - name: "spo2", - collect: async (t) => - (await fetchDailySpo2(t, query)).flatMap((s) => - toUpsert(mapDailySpo2(s, tz), "spo2", s.day), - ), - }, - { - name: "vo2max", - collect: async (t) => - (await fetchVo2Max(t, query)).flatMap((v) => - toUpsert(mapVo2Max(v, tz), "vo2max", v.timestamp ? undefined : v.day), - ), - }, - { - name: "cardio_age", - collect: async (t) => - (await fetchCardiovascularAge(t, query)).flatMap((c) => - toUpsert(mapCardiovascularAge(c, tz), "cardio_age", c.day), - ), - }, - { - name: "resilience", - collect: async (t) => - (await fetchResilience(t, query)).flatMap((r) => - toUpsert(mapResilience(r, tz), "resilience", r.day), - ), - }, - ]; + const userId = ctx.userId; + const { sleepSweeps, sleepRetirements } = ctx; + const collections: OuraCollectionSpec[] = OURA_COLLECTIONS.map((c) => ({ + name: c.name, + collect: ouraCollectionReader(c.name, ctx), + })); // Run a set of collections isolated from one another. A rejection is captured // per-collection, never propagated, so siblings still resolve. @@ -629,13 +692,25 @@ export async function syncUserOura( const lookbackDays = opts.lookbackDays ?? (await resolveOuraLookbackDays(userId)); + // Oura keys its daily collections by the user's calendar date, so the + // window is cut on the user's own days, and date-only records are anchored + // at local noon of their date in this zone. Heart rate is read by instant + // over a window of hours (`resolveOuraHeartRateWindow`). + const tz = await resolveUserTimezone(userId); + const today = userDayKey(new Date(), tz); + const ctx = ouraRunContext( + userId, + tz, + { startDate: shiftDateKey(today, -lookbackDays), endDate: today }, + await resolveOuraHeartRateWindow(userId), + ); + let result: OuraFetchResult; try { result = await fetchAll( - userId, + ctx, conn.accessToken, conn.refreshTokenCiphertext, - lookbackDays, conn.scopeDenied, ); } catch (err) { @@ -677,13 +752,15 @@ export async function syncUserOura( // Import everything the healthy collections returned regardless of whether a // sibling collection failed — one bad collection must not blank the source. let insertedSleepMeasuredAts: Date[] = []; - const imported = await upsertOuraMeasurements(userId, result.readings, { - onInserted: (rows) => { - insertedSleepMeasuredAts = rows - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt); - }, - }); + const imported = + (await upsertOuraMeasurements(userId, result.readings, { + onInserted: (rows) => { + insertedSleepMeasuredAts = rows + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt); + }, + })) + ctx.sideImported; + await refoldOuraTouchedDays(userId, ctx.heartRateTouched); // v1.29.x — best-effort Cycle Insights import. Fully isolated from the // measurement sync's status ledger on purpose: `daily_cycle_phases` sits @@ -695,14 +772,7 @@ export async function syncUserOura( try { const freshConn = await getOuraConnection(userId); if (freshConn) { - const cycleToday = userDayKey( - new Date(), - await resolveUserTimezone(userId), - ); - await syncUserOuraCyclePhases(userId, freshConn.accessToken, { - startDate: shiftDateKey(cycleToday, -lookbackDays), - endDate: cycleToday, - }); + await syncUserOuraCyclePhases(userId, freshConn.accessToken, ctx.query); } } catch (err) { getEvent()?.addWarning( @@ -761,6 +831,8 @@ export async function upsertOuraMeasurements( readings: OuraMeasurementUpsert[], opts: { onInserted?: (rows: InsertedMeasurementArrivalRow[]) => void; + /** False on a history backfill: old rows are not arrivals. */ + emitArrivals?: boolean; } = {}, ): Promise { if (readings.length === 0) return 0; @@ -818,9 +890,11 @@ export async function upsertOuraMeasurements( } opts.onInserted?.(insertedRows); - void emitInsertedMeasurementArrivals(userId, insertedRows, "oura").catch( - () => {}, - ); + if (opts.emitArrivals !== false) { + void emitInsertedMeasurementArrivals(userId, insertedRows, "oura").catch( + () => {}, + ); + } try { const keys = collapseToTypeDayKeys(touched); await recomputeBucketsForKeys(userId, keys); 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); + }); +});