Merge branch 'feat/v144-oura-depth' into feat/v144-integrations

This commit is contained in:
Marc-André Bombeck committed 2026-10-21 19:56:43 +02:00
commit ae601de16f
44 files changed
+3099 -214

No files matched your search

+2
View File
@@ -79547,6 +79547,8 @@ components:
- vo2max
- cardio_age
- resilience
- workout
- heartrate
hasToken:
description: Nightscout only.
type: boolean
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "VO2max",
"cardio_age": "kardiovaskuläres Alter",
"resilience": "Resilienz"
"resilience": "Resilienz",
"workout": "Trainings",
"heartrate": "Herzfrequenz"
}
}
}
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "VO2 max",
"cardio_age": "cardiovascular age",
"resilience": "resilience"
"resilience": "resilience",
"workout": "workouts",
"heartrate": "heart rate"
}
}
}
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "VO2 máx.",
"cardio_age": "edad cardiovascular",
"resilience": "resiliencia"
"resilience": "resiliencia",
"workout": "entrenamientos",
"heartrate": "frecuencia cardiaca"
}
}
}
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "la VO2 max",
"cardio_age": "l’âge cardiovasculaire",
"resilience": "la résilience"
"resilience": "la résilience",
"workout": "les entraînements",
"heartrate": "la fréquence cardiaque"
}
}
}
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "VO2 max",
"cardio_age": "età cardiovascolare",
"resilience": "resilienza"
"resilience": "resilienza",
"workout": "allenamenti",
"heartrate": "frequenza cardiaca"
}
}
}
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "VO2 max",
"cardio_age": "심혈관 나이",
"resilience": "회복탄력성"
"resilience": "회복탄력성",
"workout": "운동",
"heartrate": "심박수"
}
}
}
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "VO2 max",
"cardio_age": "cardiovasculaire leeftijd",
"resilience": "veerkracht"
"resilience": "veerkracht",
"workout": "trainingen",
"heartrate": "hartslag"
}
}
}
+3 -1
View File
@@ -6672,7 +6672,9 @@
"items": {
"vo2max": "VO2 max",
"cardio_age": "wiek sercowo-naczyniowy",
"resilience": "odporność"
"resilience": "odporność",
"workout": "treningi",
"heartrate": "tętno"
}
}
}
@@ -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);
+15 -5
View File
@@ -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
@@ -106,6 +106,8 @@ const BACKFILL_STAMPS: Record<string, string> = {
"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":
@@ -62,6 +62,7 @@ const WRITERS: Record<string, Disposition> = {
"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.
@@ -53,6 +53,8 @@ const WRITE_SITES: Record<string, string> = {
"lib/whoop/webhook-handler.ts": TAIL,
// Retiring the rows of Oura sleep periods the person rejected or deleted.
"lib/oura/sync.ts": TAIL,
// Retiring the equivalent-distance rows earlier Oura syncs wrote.
"lib/oura/retire-distance.ts": TAIL,
// Integration syncs: each collects the (type, day) pairs it touched and
// runs both legs itself at the end of the sync.
"lib/fitbit/sync-core.ts":
@@ -61,6 +63,8 @@ const WRITE_SITES: Record<string, string> = {
"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":
@@ -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",
@@ -218,12 +218,17 @@ describe("GET /api/oura/callback — issuer", () => {
* card for up to an hour.
*/
describe("GET /api/oura/callback — connect-time first sync", () => {
it("enqueues exactly one targeted oura-sync job for the connecting user", async () => {
it("enqueues one targeted oura-sync job and the history backfill for the connecting user", async () => {
const res = await GET(makeReq({ code: "c", state: "S", cookie: "S" }));
expect(res.headers.get("location")).toContain("oura=connected");
expect(bossSend).toHaveBeenCalledTimes(1);
expect(bossSend).toHaveBeenCalledTimes(2);
expect(bossSend).toHaveBeenCalledWith("oura-sync", { userId: "u1" });
expect(bossSend).toHaveBeenCalledWith(
"oura-backfill",
expect.objectContaining({ userId: "u1" }),
expect.objectContaining({ singletonKey: "oura-backfill|u1" }),
);
});
it("still connects when no boss instance is available", async () => {
+24
View File
@@ -18,6 +18,11 @@ import { ouraCollectionsMissingFromGrant } from "@/lib/oura/scope";
import { markReconnected } from "@/lib/integrations/status";
import { OURA_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
import {
OURA_BACKFILL_QUEUE,
type OuraBackfillPayload,
} from "@/lib/jobs/oura-backfill";
import { NextRequest, NextResponse } from "next/server";
/**
@@ -143,6 +148,10 @@ async function completeCallback(
ouraRefreshTokenEncrypted: encrypt(tokens.refresh_token),
ouraScopeDenied:
missing && missing.length > 0 ? missing : Prisma.DbNull,
// Open the history backfill again: a first connect has none yet,
// and a reconnect may grant what the old grant refused. What was
// already walked stays in the progress note and is not read again.
ouraBackfillCompletedAt: null,
},
});
@@ -165,6 +174,21 @@ async function completeCallback(
.catch((err) =>
getEvent()?.addWarning(`oura-sync connect enqueue failed: ${err}`),
);
const backfill: OuraBackfillPayload = {
userId,
enqueuedAt: new Date().toISOString(),
};
await boss
.send(
OURA_BACKFILL_QUEUE,
backfill,
integrationBackfillSourceOptions(`oura-backfill|${userId}`),
)
.catch((err) =>
getEvent()?.addWarning(
`oura-backfill connect enqueue failed: ${err}`,
),
);
}
const response = NextResponse.redirect(
@@ -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":
+2
View File
@@ -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,
+4
View File
@@ -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",
@@ -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(
@@ -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<unknown[]>
>,
}));
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<typeof import("@/lib/oura/sync-heart-rate")>()),
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<typeof import("@/lib/oura/sync")>();
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<string, { before: string }>;
};
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<string, { done?: boolean; before: string }>;
};
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<string, { denied?: boolean }>;
};
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<string, { done?: boolean }>;
};
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();
});
});
@@ -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"],
@@ -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 =
@@ -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;
}
}
}
+516
View File
@@ -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<Record<OuraCollectionName, number>> = {
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<void> {
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<boolean> {
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),
};
}
}
+1
View File
@@ -130,6 +130,7 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
"oura-sync": cohortSync("Oura"),
"strava-sync": cohortSync("Strava"),
"strava-backfill": forwarder,
"oura-backfill": forwarder,
// ── Status and insights ────────────────────────────────────────────
"insights-general-status": {
@@ -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<OuraBackfillPayload>(
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<void> {
);
}
// 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.
+1
View File
@@ -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
@@ -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);
});
});
+10 -8
View File
@@ -77,7 +77,9 @@ describe("getAuthorizationUrl", () => {
it("builds the Oura authorize URL", () => {
const url = getAuthorizationUrl("signed", CREDS);
expect(url).toContain("cloud.ouraring.com/oauth/authorize");
expect(url).toContain("scope=daily+spo2+heart_health+stress+personal");
expect(url).toContain(
"scope=daily+spo2+heart_health+stress+workout+heartrate+personal",
);
expect(url).toContain("state=signed");
});
});
@@ -471,7 +473,7 @@ describe("mapSleep", () => {
});
describe("mapDailyActivity", () => {
it("maps steps, active energy, and equivalent walking distance (m)", () => {
it("maps steps and active energy, and never the equivalent walking distance", () => {
const a: OuraDailyActivity = {
id: "1",
day: "2026-06-10",
@@ -484,12 +486,12 @@ describe("mapDailyActivity", () => {
expect(mapped.find((m) => m.type === "ACTIVE_ENERGY_BURNED")?.value).toBe(
412,
);
const dist = mapped.find((m) => m.type === "WALKING_RUNNING_DISTANCE");
expect(dist).toMatchObject({
value: 6543,
unit: "m",
fieldTag: "distance",
});
// An activity equivalent, not distance travelled: no distance row at all.
expect(mapped.map((m) => m.type).sort()).toEqual([
"ACTIVE_ENERGY_BURNED",
"ACTIVITY_STEPS",
]);
expect(mapped.some((m) => m.fieldTag === "distance")).toBe(false);
});
});
+12 -1
View File
@@ -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([]);
});
});
+63 -4
View File
@@ -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,
);
});
});
@@ -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,
);
});
});
+141
View File
@@ -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<Record<OuraCollectionName, OuraBackfillCollectionState>>;
}
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<string, unknown>;
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<string, unknown>)
: {};
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];
}),
),
};
}
+201 -44
View File
@@ -23,6 +23,7 @@ import { OuraApiError, classifyOuraResponse } from "./response-classifier";
import { envOr, envValue } from "@/lib/env";
import { importValue } from "@/lib/integrations/import-value";
import { walkPages } from "@/lib/integrations/page-walk";
import { retryAfterMsOf } from "@/lib/integrations/oauth-kit/retry-after";
import { OURA_OAUTH_SCOPE, parseOuraMissingScope } from "./scope";
const OURA_API_BASE = "https://api.ouraring.com";
@@ -449,7 +450,11 @@ export interface OuraDailyActivity {
steps?: number | null;
active_calories?: number | null;
total_calories?: number | null;
/** Equivalent walking distance for the day, metres. */
/**
* Oura's activity-equivalent distance, metres: what a walk of the day's
* effort would have covered. Not distance travelled; never imported (see
* `mapDailyActivity`).
*/
equivalent_walking_distance?: number | null;
}
@@ -459,6 +464,13 @@ interface DateRangeQuery {
maxPages?: number;
}
/** A time-range read (`heartrate`): instants, not calendar dates. */
export interface DateTimeRangeQuery {
start: Date;
end: Date;
maxPages?: number;
}
/**
* The error for a refused collection read. Oura's status alone does not say
* what went wrong; its body does (see `./scope`):
@@ -469,13 +481,16 @@ interface DateRangeQuery {
* - any other 401: the token is dead or expired (refresh, then reconnect).
* - 403: a lapsed Oura membership, per the API reference. A reconnect does
* not fix that, so it is a persistent failure rather than a reauth.
* - 429 (and 503): transient, with the `Retry-After` delay on
* `retryAfterMs` so a backfill can wait it out instead of guessing.
*/
function collectionRefusal(
verb: string,
status: number,
res: Pick<Response, "status" | "headers">,
verdict: ReturnType<typeof classifyOuraResponse>,
body: unknown,
): OuraApiError {
const status = res.status;
if (status === 401) {
const missingScope = parseOuraMissingScope(body);
if (missingScope) {
@@ -501,9 +516,43 @@ function collectionRefusal(
classification: verdict.classification,
httpStatus: verdict.httpStatus,
reason: verdict.reason,
...(status === 429 || status === 503
? { retryAfterMs: retryAfterMsOf(res) }
: {}),
});
}
/** Read one page of a collection; throws the classified refusal. */
async function readCollectionPage<T>(
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<T> | 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<T>(
path: string,
accessToken: string,
@@ -513,42 +562,61 @@ async function fetchCollection<T>(
const walk = await walkPages<string, T>(
`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<T> | 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<T>(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<T>(
path: string,
accessToken: string,
verb: string,
query: DateTimeRangeQuery,
onPage: (items: T[]) => Promise<void>,
): Promise<OuraStreamedWalk> {
const walk = await walkPages<string, never>(
`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<T>(
path,
accessToken,
verb,
params,
page,
);
await onPage(items);
return { items: [], next };
},
);
return { pages: walk.pages, truncated: walk.truncated };
}
export function fetchSleep(
accessToken: string,
query: DateRangeQuery,
@@ -665,6 +733,68 @@ export function fetchDailyCyclePhases(
);
}
/**
* Oura `workout` — one activity session, auto-detected or entered. `activity`
* is Oura's own lowercase label (`walking`, `cycling`, `strength_training`,
* ...); `distance` is metres actually covered (a GPS or step-derived figure,
* unlike the daily activity-equivalent distance); `calories` is kcal.
*/
export interface OuraWorkout {
id: string;
day?: string;
activity?: string | null;
calories?: number | null;
distance?: number | null;
start_datetime: string;
end_datetime: string;
/** `easy` | `moderate` | `hard` */
intensity?: string | null;
/** `manual` | `autodetected` | `confirmed` | `workout_heart_rate` */
source?: string | null;
/** A name the person gave the workout. Free text; never imported. */
label?: string | null;
}
export function fetchWorkouts(
accessToken: string,
query: DateRangeQuery,
): Promise<OuraWorkout[]> {
return fetchCollection<OuraWorkout>(
"/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<void>,
): Promise<OuraStreamedWalk> {
return streamCollection<OuraHeartRateSample>(
"/v2/usercollection/heartrate",
accessToken,
"fetchHeartRate",
query,
onPage,
);
}
// ─── Field → Measurement mapping ───────────────────────────────
const SEC_TO_MIN = 1 / 60;
@@ -972,8 +1102,17 @@ export function mapDailySpo2(
];
}
/** Map one Oura daily-activity record: steps, active energy, and the
* equivalent walking distance (B5 — the OURA distance ladder slot was dead). */
/**
* Map one Oura daily-activity record: steps and active energy.
*
* `equivalent_walking_distance` is deliberately not mapped. Oura derives it
* from the day's whole activity (the distance a walk of the same effort would
* have covered), so an hour of cycling or yoga adds kilometres nobody walked.
* Stored as `WALKING_RUNNING_DISTANCE` it read as real distance, and on a day
* no other source covered it became the day's distance. The rows earlier
* syncs wrote under `activity:<day>:distance` are retired on their own
* (`./retire-distance`).
*/
export function mapDailyActivity(
a: OuraDailyActivity,
tz?: string,
@@ -999,20 +1138,6 @@ export function mapDailyActivity(
fieldTag: "active_energy",
});
}
if (
typeof a.equivalent_walking_distance === "number" &&
a.equivalent_walking_distance >= 0
) {
out.push({
type: "WALKING_RUNNING_DISTANCE",
// Oura reports the equivalent walking distance in metres — the canonical
// WALKING_RUNNING_DISTANCE unit (no conversion).
value: importValue(a.equivalent_walking_distance),
unit: "m",
measuredAt,
fieldTag: "distance",
});
}
return out;
}
@@ -1106,6 +1231,38 @@ export function mapResilience(
];
}
/** Heart-rate readings outside this band are not a ring's reading. */
const OURA_BPM_MIN = 20;
const OURA_BPM_MAX = 250;
/**
* Map one `heartrate` sample → `PULSE` (bpm), the intraday heart-rate type
* Google Health and Fitbit write their minute stream to. Keyed on the
* sample's own instant (`heartrate:<ISO>`): 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
+9 -1
View File
@@ -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;
}
}
+72
View File
@@ -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:<day>: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<number> {
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;
}
+6
View File
@@ -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. */
+133
View File
@@ -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<readonly [string, WorkoutSportType]> = [
// 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<string, WorkoutSportType> = 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";
}
+200
View File
@@ -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:<ISO>`), 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<number> {
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<void> {
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,
};
}
+150
View File
@@ -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<number> {
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<number> {
const rows: OuraWorkoutRow[] = [];
for (const w of workouts) {
const row = mapOuraWorkout(w);
if (row) rows.push(row);
}
return upsertOuraWorkouts(userId, rows, opts);
}
+214 -140
View File
@@ -4,7 +4,7 @@
* Pulls daily readiness (→ RECOVERY_SCORE + BODY_TEMPERATURE_DEVIATION), sleep
* (real per-segment hypnogram timeline when present, else per-stage totals;
* efficiency, HRV, RHR, respiratory rate), daily activity (steps, active
* energy, equivalent walking distance), the daily Sleep Score (→ SLEEP_SCORE),
* energy), the daily Sleep Score (→ SLEEP_SCORE),
* daily SpO2 (→ OXYGEN_SATURATION), the dedicated vO2_max collection (→
* VO2_MAX), and the daily cardiovascular-age estimate (→ VASCULAR_AGE) for one
* connected user, mapping each into `Measurement` rows tagged `source = OURA`.
@@ -75,6 +75,7 @@ import {
fetchResilience,
fetchSleep,
fetchVo2Max,
fetchWorkouts,
isRetiredOuraSleepPeriod,
mapCardiovascularAge,
mapDailyActivity,
@@ -90,7 +91,18 @@ import { getOuraConnection, syncOuraScopeDenied } from "./credentials";
import { refreshOuraAfterRefusal } from "./oauth-descriptor";
import { OuraApiError, classifyOuraError } from "./response-classifier";
import { syncUserOuraCyclePhases } from "./cycle-sync";
import { isOuraScopeGatedCollection } from "./scope";
import {
OURA_COLLECTIONS,
isOuraScopeGatedCollection,
type OuraCollectionName,
} from "./scope";
import { writeOuraWorkouts } from "./sync-workouts";
import {
refoldOuraTouchedDays,
resolveOuraHeartRateWindow,
syncOuraHeartRateRange,
TouchedDays,
} from "./sync-heart-rate";
import { shiftDateKey, userDayKey } from "@/lib/tz/format";
import { resolveUserTimezone } from "@/lib/tz/resolver";
@@ -220,9 +232,12 @@ function toUpsert(
/** One Oura collection: its ledger name and a self-contained fetch+map closure
* that resolves to the collection's mapped upserts. Each closure owns its own
* client call and mapper so a throw is contained to this one collection. */
* client call and mapper so a throw is contained to this one collection. A
* collection that writes rows of its own (workouts, the heart-rate stream)
* writes inside the closure, counts on `OuraRunContext.sideImported`, and
* resolves to no upserts. */
interface OuraCollectionSpec {
name: string;
name: OuraCollectionName;
collect: (token: string) => Promise<OuraMeasurementUpsert[]>;
}
@@ -261,12 +276,167 @@ export interface OuraFetchResult {
scopeDenied: string[] | null;
}
/**
* What one run reads: the day window of the day-keyed collections, the time
* range of the heart-rate stream, the user's zone, and the sinks the sleep
* collection and the self-writing collections fill as they go.
*/
export interface OuraRunContext {
userId: string;
tz: string;
query: { startDate: string; endDate: string };
heartRate: { start: Date; end: Date };
/** False on a history backfill: old rows are not arrivals. */
emitArrivals: boolean;
sleepSweeps: SleepSegmentSweep[];
sleepRetirements: string[];
/** Rows written by the collections that write their own. */
sideImported: number;
/** Days the heart-rate stream touched; refolded once by the caller. */
heartRateTouched: TouchedDays;
}
export function ouraRunContext(
userId: string,
tz: string,
query: { startDate: string; endDate: string },
heartRate: { start: Date; end: Date },
emitArrivals = true,
): OuraRunContext {
return {
userId,
tz,
query,
heartRate,
emitArrivals,
sleepSweeps: [],
sleepRetirements: [],
sideImported: 0,
heartRateTouched: new TouchedDays(),
};
}
/**
* The fetch+map (or fetch+write) closure of one collection over the context's
* window. The one place each collection's read is defined, for the hourly
* sync and the history backfill alike.
*/
export function ouraCollectionReader(
name: OuraCollectionName,
ctx: OuraRunContext,
): (token: string) => Promise<OuraMeasurementUpsert[]> {
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:<id>:` (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:<i>` rows, and a
// stale stage-total set once the hypnogram appears.
ctx.sleepSweeps.push({
prefix: `sleep:${s.id}:`,
keepIds: rows
.filter((r) => r.type === "SLEEP_DURATION")
.map((r) => r.externalId),
});
}
return ups;
};
case "activity":
return async (t) =>
(await fetchDailyActivity(t, query)).flatMap((a) =>
toUpsert(
mapDailyActivity(a, tz),
"activity",
a.timestamp ? undefined : a.day,
),
);
case "daily_sleep":
return async (t) =>
(await fetchDailySleep(t, query)).flatMap((d) =>
toUpsert(
mapDailySleep(d, tz),
"daily_sleep",
d.timestamp ? undefined : d.day,
),
);
case "spo2":
return async (t) =>
(await fetchDailySpo2(t, query)).flatMap((s) =>
toUpsert(mapDailySpo2(s, tz), "spo2", s.day),
);
case "vo2max":
return async (t) =>
(await fetchVo2Max(t, query)).flatMap((v) =>
toUpsert(mapVo2Max(v, tz), "vo2max", v.timestamp ? undefined : v.day),
);
case "cardio_age":
return async (t) =>
(await fetchCardiovascularAge(t, query)).flatMap((c) =>
toUpsert(mapCardiovascularAge(c, tz), "cardio_age", c.day),
);
case "resilience":
return async (t) =>
(await fetchResilience(t, query)).flatMap((r) =>
toUpsert(mapResilience(r, tz), "resilience", r.day),
);
case "workout":
return async (t) => {
const workouts = await fetchWorkouts(t, query);
ctx.sideImported += await writeOuraWorkouts(ctx.userId, workouts, {
emitArrivals: ctx.emitArrivals,
});
return [];
};
case "heartrate":
return async (t) => {
const { imported, truncated } = await syncOuraHeartRateRange(
ctx.userId,
t,
ctx.heartRate,
{ emitArrivals: ctx.emitArrivals, touched: ctx.heartRateTouched },
);
ctx.sideImported += imported;
if (truncated) {
// A window Oura could not page to its end is not read: fail it so
// the run is not reported clean over a gap.
throw new Error("oura heart-rate walk stopped short of its window");
}
return [];
};
}
}
/**
* A 401 whose body names the missing scope (see `./scope`): Oura checked the
* token and found it valid, the grant just does not cover this collection. A
* refresh cannot widen a grant, so none is spent on it.
*/
function isNamedScopeRefusal(err: unknown): boolean {
export function isNamedScopeRefusal(err: unknown): boolean {
return (
err instanceof OuraApiError &&
err.httpStatus === 401 &&
@@ -279,7 +449,7 @@ function isNamedScopeRefusal(err: unknown): boolean {
* no body to tell. This is what the one refresh is for. A 403 is neither: Oura
* answers it for a lapsed membership, which no refresh or reconnect fixes.
*/
function isTokenRefusal(err: unknown): boolean {
export function isTokenRefusal(err: unknown): boolean {
return (
err instanceof OuraApiError &&
err.httpStatus === 401 &&
@@ -346,124 +516,17 @@ function splitScopeDenied(
* connection at `reauth_required`.
*/
async function fetchAll(
userId: string,
ctx: OuraRunContext,
accessToken: string,
refreshTokenCiphertext: string,
lookbackDays: number,
knownScopeDenied: readonly string[] = [],
): Promise<OuraFetchResult> {
// 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:<id>:` (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:<i>` 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<number> {
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);
+502
View File
@@ -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<string>(),
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<typeof import("@/lib/safe-fetch")>();
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<void> {
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<string, unknown> })
.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);
});
});