diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index e452322f1..2d4f52a2e 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/src/lib/oura/__tests__/sync.test.ts @@ -338,7 +338,79 @@ describe("syncUserOura", () => { reason: "http_403", }), ); - await expect(syncUserOura("u1")).rejects.toBeInstanceOf(OuraApiError); + // A 403 is a per-collection hard failure now (isolated, not a whole-batch + // reject): it records a partial failure and does NOT trigger a refresh. + await syncUserOura("u1"); + expect(refreshMock).not.toHaveBeenCalled(); + expect(recordFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ integration: "oura" }), + ); + expect(recordSuccessMock).not.toHaveBeenCalled(); + }); + + it("isolates one bad collection — siblings still import (F-SYNC-2)", async () => { + getConnMock.mockResolvedValue(CONN); + // Readiness endpoint is flaky (500), but sleep-score + spo2 return data. + fetchReadinessMock.mockRejectedValue( + new OuraApiError({ + verb: "fetchReadiness", + classification: "transient", + httpStatus: 500, + reason: "http_500", + }), + ); + fetchDailySleepMock.mockResolvedValue([ + { id: "s", day: "2026-06-10", score: 80 }, + ]); + fetchSpo2Mock.mockResolvedValue([ + { id: "o", day: "2026-06-10", spo2_percentage: { average: 97 } }, + ]); + + const imported = await syncUserOura("u1"); + + // The healthy collections still wrote their rows — the source is not blanked. + expect(imported).toBe(2); + const written = upsertMock.mock.calls.map( + (c) => c[0].where.userId_type_source_externalId.type, + ); + expect(written).toContain("SLEEP_SCORE"); + expect(written).toContain("OXYGEN_SATURATION"); + // The cycle is still marked failed (partial) so freshness stays honest. + expect(recordFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ integration: "oura" }), + ); + expect(recordSuccessMock).not.toHaveBeenCalled(); expect(refreshMock).not.toHaveBeenCalled(); }); + + it("isolates a throwing mapper — one malformed record does not blank siblings (F-SYNC-2)", async () => { + getConnMock.mockResolvedValue(CONN); + // A malformed readiness record whose mapper throws must not reject the + // whole batch (the old Promise.all failure mode). + fetchReadinessMock.mockReturnValue( + Promise.resolve([ + new Proxy( + {}, + { + get() { + throw new Error("malformed readiness point"); + }, + }, + ), + ]), + ); + fetchSpo2Mock.mockResolvedValue([ + { id: "o", day: "2026-06-10", spo2_percentage: { average: 96 } }, + ]); + + const imported = await syncUserOura("u1"); + + expect(imported).toBe(1); + const written = upsertMock.mock.calls.map( + (c) => c[0].where.userId_type_source_externalId.type, + ); + expect(written).toContain("OXYGEN_SATURATION"); + expect(recordFailureMock).toHaveBeenCalled(); + expect(recordSuccessMock).not.toHaveBeenCalled(); + }); }); diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index 363b3c485..5c584433a 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -114,11 +114,42 @@ 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. */ +interface OuraCollectionSpec { + name: string; + collect: (token: string) => Promise; +} + +/** A single collection that failed to fetch/map, carried so the caller can + * classify + record the failure without blanking the collections that did + * succeed. */ +export interface OuraCollectionFailure { + name: string; + err: unknown; +} + +export interface OuraFetchResult { + readings: OuraMeasurementUpsert[]; + failures: OuraCollectionFailure[]; +} + /** * Fetch every Oura daily collection for a user with a single reactive - * refresh-on-401 retry. Returns the raw mapped readings (not yet upserted). - * Throws a classified `OuraApiError` on a hard failure so the caller records - * the ledger entry. + * refresh-on-401 retry, ISOLATED per collection. Unlike a bare `Promise.all`, + * one flaky endpoint or one throwing mapper no longer rejects the whole batch + * (which used to blank readiness, sleep, activity, spo2 all at once). Each + * collection is settled independently (`Promise.allSettled`); the ones that + * succeed import, the ones that throw are returned as `failures` for the caller + * to record — mirroring Google Health's / Fitbit's per-collection hard-fail + * ledger. + * + * The reactive refresh is preserved: because every collection shares one access + * token, an expired token 401s all of them together; a single refresh is done + * once and ONLY the failed collections are retried with the rotated token. A + * failed refresh (`invalid_grant`) still throws so the caller parks the whole + * connection at `reauth_required`. */ async function fetchAll( userId: string, @@ -126,72 +157,131 @@ async function fetchAll( refreshToken: string, refreshTokenCiphertext: string, lookbackDays: number, -): Promise { +): Promise { const now = new Date(); const start = new Date(now.getTime() - lookbackDays * 24 * 60 * 60 * 1000); const query = { startDate: ymd(start), endDate: ymd(now) }; - const run = async (token: string): Promise => { - const [ - readiness, - sleeps, - activities, - dailySleep, - spo2, - vo2max, - cardioAge, - resilience, - ] = await Promise.all([ - fetchReadiness(token, query), - fetchSleep(token, query), - fetchDailyActivity(token, query), - fetchDailySleep(token, query), - fetchDailySpo2(token, query), - fetchVo2Max(token, query), - fetchCardiovascularAge(token, query), - fetchResilience(token, query), - ]); - const out: OuraMeasurementUpsert[] = []; - for (const r of readiness) - out.push(...toUpsert(mapReadiness(r), "readiness")); - for (const s of sleeps) out.push(...toUpsert(mapSleep(s), "sleep")); - for (const a of activities) - out.push(...toUpsert(mapDailyActivity(a), "activity")); - for (const d of dailySleep) - out.push(...toUpsert(mapDailySleep(d), "daily_sleep")); - for (const s of spo2) out.push(...toUpsert(mapDailySpo2(s), "spo2")); - for (const v of vo2max) out.push(...toUpsert(mapVo2Max(v), "vo2max")); - for (const c of cardioAge) - out.push(...toUpsert(mapCardiovascularAge(c), "cardio_age")); - for (const r of resilience) - out.push(...toUpsert(mapResilience(r), "resilience")); - return out; + const collections: OuraCollectionSpec[] = [ + { + name: "readiness", + collect: async (t) => + (await fetchReadiness(t, query)).flatMap((r) => + toUpsert(mapReadiness(r), "readiness"), + ), + }, + { + name: "sleep", + collect: async (t) => + (await fetchSleep(t, query)).flatMap((s) => + toUpsert(mapSleep(s), "sleep"), + ), + }, + { + name: "activity", + collect: async (t) => + (await fetchDailyActivity(t, query)).flatMap((a) => + toUpsert(mapDailyActivity(a), "activity"), + ), + }, + { + name: "daily_sleep", + collect: async (t) => + (await fetchDailySleep(t, query)).flatMap((d) => + toUpsert(mapDailySleep(d), "daily_sleep"), + ), + }, + { + name: "spo2", + collect: async (t) => + (await fetchDailySpo2(t, query)).flatMap((s) => + toUpsert(mapDailySpo2(s), "spo2"), + ), + }, + { + name: "vo2max", + collect: async (t) => + (await fetchVo2Max(t, query)).flatMap((v) => + toUpsert(mapVo2Max(v), "vo2max"), + ), + }, + { + name: "cardio_age", + collect: async (t) => + (await fetchCardiovascularAge(t, query)).flatMap((c) => + toUpsert(mapCardiovascularAge(c), "cardio_age"), + ), + }, + { + name: "resilience", + collect: async (t) => + (await fetchResilience(t, query)).flatMap((r) => + toUpsert(mapResilience(r), "resilience"), + ), + }, + ]; + + // Run a set of collections isolated from one another. A rejection is captured + // per-collection, never propagated, so siblings still resolve. + const attempt = async ( + token: string, + specs: OuraCollectionSpec[], + ): Promise<{ + readings: OuraMeasurementUpsert[]; + failed: OuraCollectionSpec[]; + failures: OuraCollectionFailure[]; + auth401: boolean; + }> => { + const settled = await Promise.allSettled( + specs.map((spec) => spec.collect(token)), + ); + const readings: OuraMeasurementUpsert[] = []; + const failed: OuraCollectionSpec[] = []; + const failures: OuraCollectionFailure[] = []; + let auth401 = false; + settled.forEach((res, i) => { + const spec = specs[i]!; + if (res.status === "fulfilled") { + readings.push(...res.value); + } else { + const err = res.reason; + if (err instanceof OuraApiError && err.httpStatus === 401) + auth401 = true; + failed.push(spec); + failures.push({ name: spec.name, err }); + } + }); + return { readings, failed, failures, auth401 }; }; - try { - return await run(accessToken); - } catch (err) { - // Reactive refresh: a 401 means the access token expired. Refresh once - // (rotating both tokens) and retry. A 403 / other error is NOT a refresh - // case — rethrow so the caller classifies it. - if (!(err instanceof OuraApiError) || err.httpStatus !== 401) throw err; - - const creds = await getOuraClientCredentials(userId); - if (!creds) throw err; - - const rotated = await refreshAccessToken(refreshToken, creds); - // Compare-and-swap persist: on a lost race against a concurrent sync this - // returns the peer's freshly rotated access token rather than the (now - // invalidated) one we just minted, so neither sync parks the connection. - const usableToken = await storeOuraTokens( - userId, - rotated.access_token, - rotated.refresh_token, - refreshTokenCiphertext, - ); - if (!usableToken) throw err; - return run(usableToken); + const first = await attempt(accessToken, collections); + if (!first.auth401) { + return { readings: first.readings, failures: first.failures }; } + + // Reactive refresh: the access token expired (a 401 on the shared token). + // Refresh once (rotating both tokens) and retry ONLY the collections that + // failed. A failed refresh throws so the caller parks reauth_required. + const creds = await getOuraClientCredentials(userId); + if (!creds) return { readings: first.readings, failures: first.failures }; + + const rotated = await refreshAccessToken(refreshToken, creds); + // Compare-and-swap persist: on a lost race against a concurrent sync this + // returns the peer's freshly rotated access token rather than the (now + // invalidated) one we just minted, so neither sync parks the connection. + const usableToken = await storeOuraTokens( + userId, + rotated.access_token, + rotated.refresh_token, + refreshTokenCiphertext, + ); + if (!usableToken) return { readings: first.readings, failures: first.failures }; + + const retry = await attempt(usableToken, first.failed); + return { + readings: [...first.readings, ...retry.readings], + failures: retry.failures, + }; } /** @@ -206,9 +296,9 @@ export async function syncUserOura( const conn = await getOuraConnection(userId); if (!conn) return 0; - let readings: OuraMeasurementUpsert[]; + let result: OuraFetchResult; try { - readings = await fetchAll( + result = await fetchAll( userId, conn.accessToken, conn.refreshToken, @@ -216,6 +306,8 @@ export async function syncUserOura( opts.lookbackDays ?? OURA_SYNC_LOOKBACK_DAYS, ); } catch (err) { + // Only a whole-connection failure (a failed token refresh → reauth) reaches + // here now; per-collection failures are captured on `result.failures`. await recordSyncFailure({ userId, integration: "oura", @@ -229,7 +321,37 @@ export async function syncUserOura( throw err; } - const imported = await upsertOuraMeasurements(userId, readings); + // Import everything the healthy collections returned regardless of whether a + // sibling collection failed — one bad collection must not blank the source. + const imported = await upsertOuraMeasurements(userId, result.readings); + + if (result.failures.length > 0) { + // Partial failure: keep the cycle honest. Record the failure and do NOT + // stamp success, so the freshness surface reflects that some collections + // are behind and the next tick refetches them, rather than showing green. + const firstErr = result.failures[0]!.err; + getEvent()?.addWarning( + `oura: ${result.failures.length} collection(s) failed for ${userId}: ${result.failures + .map((f) => f.name) + .join(", ")}`, + ); + await recordSyncFailure({ + userId, + integration: "oura", + kind: classifyOuraFailure(firstErr), + message: `partial sync failure (${result.failures + .map((f) => f.name) + .join(", ")}): ${ + firstErr instanceof Error ? firstErr.message : String(firstErr) + }`, + errorCode: + firstErr instanceof OuraApiError && firstErr.httpStatus != null + ? String(firstErr.httpStatus) + : undefined, + }); + return imported; + } + await recordSyncSuccess(userId, "oura"); return imported; }