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);