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