mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-07 04:12:04 +00:00
feat(oura): backfill the full history in resumable slices
Every Oura collection, workouts and heart rate included, is now read back to the start of the account's history by a backfill that runs through the shared integration lane in five-minute slices. Each collection walks from the day the backfill began toward the past one window at a time (ninety days, thirty for sleep, seven for heart rate) and the progress note on the user row records after every window where it got to, so a slice cut short by its budget, the lane or a rate limit resumes there. A 429 ends the slice and the next one waits for the Retry-After Oura named. Activity and sleep are walked first to learn where history starts; the rest stop a week before it. Heart rate streams a page at a time and keeps only the days it touched, so memory stays flat however long the history is. A collection the grant does not cover is a scope note, not a failure; the first slice of the next chain asks again, and a connect clears the completion stamp so a newly granted scope is read without walking anything twice. The two columns arrive empty, so every existing connection is offered the backfill at the next boot without reconnecting. Before anything else the backfill retires the WALKING_RUNNING_DISTANCE rows earlier syncs wrote from the equivalent walking distance and refolds their days; an account that kept such rows after disconnecting is offered too.
This commit is contained in:
Marc-André Bombeck committed
1 parent
7585994697
commit
bce9ad1c25
21 files changed
+1462
-10
No files matched your search
@@ -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
@@ -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":
|
||||
|
||||
@@ -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":
|
||||
|
||||
@@ -206,12 +206,17 @@ describe("GET /api/oura/callback — issuer", () => {
|
||||
* card for up to an hour.
|
||||
*/
|
||||
describe("GET /api/oura/callback — connect-time first sync", () => {
|
||||
it("enqueues exactly one targeted oura-sync job for the connecting user", async () => {
|
||||
it("enqueues one targeted oura-sync job and the history backfill for the connecting user", async () => {
|
||||
const res = await GET(makeReq({ code: "c", state: "S", cookie: "S" }));
|
||||
|
||||
expect(res.headers.get("location")).toContain("oura=connected");
|
||||
expect(bossSend).toHaveBeenCalledTimes(1);
|
||||
expect(bossSend).toHaveBeenCalledTimes(2);
|
||||
expect(bossSend).toHaveBeenCalledWith("oura-sync", { userId: "u1" });
|
||||
expect(bossSend).toHaveBeenCalledWith(
|
||||
"oura-backfill",
|
||||
expect.objectContaining({ userId: "u1" }),
|
||||
expect.objectContaining({ singletonKey: "oura-backfill|u1" }),
|
||||
);
|
||||
});
|
||||
|
||||
it("still connects when no boss instance is available", async () => {
|
||||
|
||||
@@ -17,6 +17,11 @@ import { ouraCollectionsMissingFromGrant } from "@/lib/oura/scope";
|
||||
import { markReconnected } from "@/lib/integrations/status";
|
||||
import { OURA_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
|
||||
import {
|
||||
OURA_BACKFILL_QUEUE,
|
||||
type OuraBackfillPayload,
|
||||
} from "@/lib/jobs/oura-backfill";
|
||||
import { NextRequest, NextResponse } from "next/server";
|
||||
|
||||
/**
|
||||
@@ -134,6 +139,10 @@ async function completeCallback(
|
||||
ouraRefreshTokenEncrypted: encrypt(tokens.refresh_token),
|
||||
ouraScopeDenied:
|
||||
missing && missing.length > 0 ? missing : Prisma.DbNull,
|
||||
// Open the history backfill again: a first connect has none yet,
|
||||
// and a reconnect may grant what the old grant refused. What was
|
||||
// already walked stays in the progress note and is not read again.
|
||||
ouraBackfillCompletedAt: null,
|
||||
},
|
||||
});
|
||||
|
||||
@@ -153,6 +162,21 @@ async function completeCallback(
|
||||
.catch((err) =>
|
||||
getEvent()?.addWarning(`oura-sync connect enqueue failed: ${err}`),
|
||||
);
|
||||
const backfill: OuraBackfillPayload = {
|
||||
userId,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
};
|
||||
await boss
|
||||
.send(
|
||||
OURA_BACKFILL_QUEUE,
|
||||
backfill,
|
||||
integrationBackfillSourceOptions(`oura-backfill|${userId}`),
|
||||
)
|
||||
.catch((err) =>
|
||||
getEvent()?.addWarning(
|
||||
`oura-backfill connect enqueue failed: ${err}`,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
const response = NextResponse.redirect(
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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];
|
||||
}),
|
||||
),
|
||||
};
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -436,7 +436,7 @@ export function ouraCollectionReader(
|
||||
* token and found it valid, the grant just does not cover this collection. A
|
||||
* refresh cannot widen a grant, so none is spent on it.
|
||||
*/
|
||||
function isNamedScopeRefusal(err: unknown): boolean {
|
||||
export function isNamedScopeRefusal(err: unknown): boolean {
|
||||
return (
|
||||
err instanceof OuraApiError &&
|
||||
err.httpStatus === 401 &&
|
||||
@@ -449,7 +449,7 @@ function isNamedScopeRefusal(err: unknown): boolean {
|
||||
* no body to tell. This is what the one refresh is for. A 403 is neither: Oura
|
||||
* answers it for a lapsed membership, which no refresh or reconnect fixes.
|
||||
*/
|
||||
function isTokenRefusal(err: unknown): boolean {
|
||||
export function isTokenRefusal(err: unknown): boolean {
|
||||
return (
|
||||
err instanceof OuraApiError &&
|
||||
err.httpStatus === 401 &&
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user