diff --git a/src/lib/jobs/reminder/__tests__/polar-sync-legs.test.ts b/src/lib/jobs/reminder/__tests__/polar-sync-legs.test.ts index 3d1c30923..557cf14e5 100644 --- a/src/lib/jobs/reminder/__tests__/polar-sync-legs.test.ts +++ b/src/lib/jobs/reminder/__tests__/polar-sync-legs.test.ts @@ -30,6 +30,18 @@ describe("syncUserPolarLegs — composite independence on the shared polar ledge expect(syncUserPolarWorkouts).toHaveBeenCalledWith("u1"); }); + it("sums both counts without throwing when the vitals leg returns a partial-failure count", async () => { + // Post-R7 the vitals leg records a partial failure INTERNALLY and RETURNS + // its healthy-collections import count rather than throwing. The wrapper + // must sum it with the workout leg and complete cleanly — a partial vitals + // failure never starves the user's success accounting at the cohort level. + syncUserPolar.mockResolvedValue(4); + syncUserPolarWorkouts.mockResolvedValue(2); + await expect(syncUserPolarLegs("u1")).resolves.toBe(6); + expect(syncUserPolar).toHaveBeenCalledWith("u1"); + expect(syncUserPolarWorkouts).toHaveBeenCalledWith("u1"); + }); + it("still attempts the workout leg when the vitals leg fails (vice-versa isolation)", async () => { syncUserPolar.mockRejectedValue(new Error("vitals down")); syncUserPolarWorkouts.mockResolvedValue(3); diff --git a/src/lib/jobs/reminder/polar-sync.ts b/src/lib/jobs/reminder/polar-sync.ts index ae06b02c9..ee7c80253 100644 --- a/src/lib/jobs/reminder/polar-sync.ts +++ b/src/lib/jobs/reminder/polar-sync.ts @@ -27,6 +27,12 @@ export type PolarSyncPayload = PollCohortPayload; * always attempted, so a workouts-leg failure can never mask a vitals success * and vice versa — then the FIRST error is rethrown so the cohort handler warns * and skips this user's success accounting without losing the error. + * + * The vitals leg (`syncUserPolar`) now settles its five collections + * independently: a fetch/map failure on one collection is recorded as a partial + * failure and the leg RETURNS (no rethrow, no success stamp), importing the + * healthy collections. So the vitals leg reaching this wrapper's catch means a + * write-path/unexpected error, not a collection fetch failure. */ export async function syncUserPolarLegs(userId: string): Promise { let total = 0; @@ -63,8 +69,10 @@ export const handlePolarSync = makePollCohortHandler({ return users.map((u) => u.id); }, syncUser: syncUserPolarLegs, - // Catches an UNMARKED escape — today the `upsertPolarMeasurements` / - // `upsertPolarWorkouts` write-path throws (the fetch paths record + mark - // their own failures, and the legs wrapper rethrows the marked error). + // Catches an UNMARKED escape — the `upsertPolarMeasurements` / + // `upsertPolarWorkouts` write-path throw. The vitals leg records its own + // per-collection partial failures inline and returns without rethrowing; the + // workouts leg records + marks its own fetch failures. Only an unmarked + // write-path/unexpected escape reaches this boundary recorder. recordFailure: recordPolarSyncFailure, }); diff --git a/src/lib/polar/__tests__/sync.test.ts b/src/lib/polar/__tests__/sync.test.ts index 71aace68c..56e33a343 100644 --- a/src/lib/polar/__tests__/sync.test.ts +++ b/src/lib/polar/__tests__/sync.test.ts @@ -331,8 +331,9 @@ describe("syncUserPolar", () => { ); }); - it("records a reauth_required failure on a 401 and rethrows", async () => { + it("records a reauth_required partial failure on a 401 and does NOT throw (per-collection isolation)", async () => { getConnMock.mockResolvedValue(CONN); + // The recharge collection 401s; the four siblings return data. fetchRechargesMock.mockRejectedValue( new PolarApiError({ verb: "fetchNightlyRecharges", @@ -341,11 +342,17 @@ describe("syncUserPolar", () => { reason: "http_401", }), ); - const err = await syncUserPolar("u1").catch((e) => e); - expect(err).toBeInstanceOf(PolarApiError); - // Recorded at the source AND marked (the legs wrapper rethrows the marked - // error), so the poll-cohort boundary does not double-record it. - expect(isSyncFailureRecorded(err)).toBe(true); + fetchCardioLoadsMock.mockResolvedValue([ + { date: "2026-06-10", cardio_load: 42 }, + ]); + + const imported = await syncUserPolar("u1"); + + // No rethrow now — a fetch failure is isolated per-collection. + expect(imported).toBe(1); + expect(createdRows().map((row) => row.type)).toContain("CARDIO_LOAD"); + // Exactly one partial failure recorded, classified off the first error, + // naming the failed collection. expect(recordFailureMock).toHaveBeenCalledTimes(1); expect(recordFailureMock).toHaveBeenCalledWith( expect.objectContaining({ @@ -353,11 +360,206 @@ describe("syncUserPolar", () => { integration: "polar", kind: "reauth_required", errorCode: "401", + message: expect.stringContaining("recharge"), }), ); expect(recordSuccessMock).not.toHaveBeenCalled(); }); + it("isolates one dead collection — imports the healthy four and records one named failure", async () => { + getConnMock.mockResolvedValue(CONN); + // SpO2 403s (no sensor / consent), the other four return one row each. + fetchSpo2Mock.mockRejectedValue( + new PolarApiError({ + verb: "fetchSpo2", + classification: "reauth_required", + httpStatus: 403, + reason: "http_403", + }), + ); + fetchRechargesMock.mockResolvedValue([ + { date: "2026-06-10", nightly_recharge_status: 6 }, + ]); + fetchSleepsMock.mockResolvedValue([ + { + date: "2026-06-10", + sleep_start_time: "2026-06-09T23:00:00+02:00", + sleep_end_time: "2026-06-10T07:00:00+02:00", + light_sleep: 3600, + }, + ]); + fetchActivitiesMock.mockResolvedValue([ + { start_time: "2026-06-10T00:00:00", steps: 8000 }, + ]); + fetchCardioLoadsMock.mockResolvedValue([ + { date: "2026-06-10", cardio_load: 55 }, + ]); + + const imported = await syncUserPolar("u1"); + + // The healthy four still wrote their rows — the source is not blanked. + const written = createdRows().map((row) => row.type); + expect(written).toContain("RECOVERY_SCORE"); + expect(written).toContain("SLEEP_DURATION"); + expect(written).toContain("ACTIVITY_STEPS"); + expect(written).toContain("CARDIO_LOAD"); + expect(written).not.toContain("OXYGEN_SATURATION"); + expect(imported).toBeGreaterThan(0); + // Exactly one failure, classified off the 403, naming spo2. No success. + expect(recordFailureMock).toHaveBeenCalledTimes(1); + expect(recordFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "polar", + kind: "reauth_required", + errorCode: "403", + message: expect.stringContaining("spo2"), + }), + ); + expect(recordSuccessMock).not.toHaveBeenCalled(); + }); + + it("isolates a throwing mapper — one malformed record does not blank siblings", async () => { + getConnMock.mockResolvedValue(CONN); + // A malformed recharge record whose mapper throws on first field access + // must not reject the whole batch (the old Promise.all failure mode). + fetchRechargesMock.mockResolvedValue([ + new Proxy( + {}, + { + get() { + throw new Error("malformed recharge point"); + }, + }, + ), + ]); + fetchCardioLoadsMock.mockResolvedValue([ + { date: "2026-06-10", cardio_load: 77 }, + ]); + + const imported = await syncUserPolar("u1"); + + expect(imported).toBe(1); + expect(createdRows().map((row) => row.type)).toContain("CARDIO_LOAD"); + expect(recordFailureMock).toHaveBeenCalledTimes(1); + expect(recordFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringContaining("recharge") }), + ); + expect(recordSuccessMock).not.toHaveBeenCalled(); + }); + + it("sleep-collection failure skips the sweep, siblings still import", async () => { + getConnMock.mockResolvedValue(CONN); + fetchSleepsMock.mockRejectedValue( + new PolarApiError({ + verb: "fetchSleeps", + classification: "transient", + httpStatus: 500, + reason: "http_500", + }), + ); + fetchActivitiesMock.mockResolvedValue([ + { start_time: "2026-06-10T00:00:00", steps: 8000 }, + ]); + + const imported = await syncUserPolar("u1"); + + // The sweep never ran — a failed sleep collection pushes no sweep entries. + expect(updateManyMock).not.toHaveBeenCalled(); + // The activity sibling still imported. + expect(createdRows().map((row) => row.type)).toContain("ACTIVITY_STEPS"); + expect(imported).toBeGreaterThan(0); + expect(recordFailureMock).toHaveBeenCalledTimes(1); + expect(recordSuccessMock).not.toHaveBeenCalled(); + }); + + it("all five healthy → success stamped once, no failure recorded", async () => { + getConnMock.mockResolvedValue(CONN); + fetchRechargesMock.mockResolvedValue([ + { date: "2026-06-10", nightly_recharge_status: 6 }, + ]); + + await syncUserPolar("u1"); + + expect(recordFailureMock).not.toHaveBeenCalled(); + expect(recordSuccessMock).toHaveBeenCalledTimes(1); + expect(recordSuccessMock).toHaveBeenCalledWith("u1", "polar"); + }); + + it.each([ + { name: "recharge", type: "RECOVERY_SCORE" }, + { name: "sleep", type: "SLEEP_DURATION" }, + { name: "activity", type: "ACTIVITY_STEPS" }, + { name: "cardioload", type: "CARDIO_LOAD" }, + { name: "spo2", type: "OXYGEN_SATURATION" }, + ])( + "Oura-parity: a 4xx on the $name collection names it in the single failure and imports every sibling", + async ({ name }) => { + getConnMock.mockResolvedValue(CONN); + // Every collection returns one healthy row... + fetchRechargesMock.mockResolvedValue([ + { date: "2026-06-10", nightly_recharge_status: 6 }, + ]); + fetchSleepsMock.mockResolvedValue([ + { + date: "2026-06-10", + sleep_start_time: "2026-06-09T23:00:00+02:00", + sleep_end_time: "2026-06-10T07:00:00+02:00", + light_sleep: 3600, + }, + ]); + fetchActivitiesMock.mockResolvedValue([ + { start_time: "2026-06-10T00:00:00", steps: 8000 }, + ]); + fetchCardioLoadsMock.mockResolvedValue([ + { date: "2026-06-10", cardio_load: 55 }, + ]); + fetchSpo2Mock.mockResolvedValue([ + { test_time: 1_781_059_600, blood_oxygen_percent: 96 }, + ]); + // ...except the one under test, which 4xxs. + const mockByName: Record> = { + recharge: fetchRechargesMock, + sleep: fetchSleepsMock, + activity: fetchActivitiesMock, + cardioload: fetchCardioLoadsMock, + spo2: fetchSpo2Mock, + }; + mockByName[name]!.mockRejectedValue( + new PolarApiError({ + verb: name, + classification: "reauth_required", + httpStatus: 403, + reason: "http_403", + }), + ); + + await syncUserPolar("u1"); + + // Exactly one failure, naming this collection; no success stamp. + expect(recordFailureMock).toHaveBeenCalledTimes(1); + expect(recordFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "polar", + message: expect.stringContaining(name), + }), + ); + expect(recordSuccessMock).not.toHaveBeenCalled(); + // Every sibling collection still imported at least one row. + const written = createdRows().map((row) => row.type); + const siblingTypes: Record = { + recharge: "RECOVERY_SCORE", + sleep: "SLEEP_DURATION", + activity: "ACTIVITY_STEPS", + cardioload: "CARDIO_LOAD", + spo2: "OXYGEN_SATURATION", + }; + for (const [collName, collType] of Object.entries(siblingTypes)) { + if (collName === name) continue; + expect(written).toContain(collType); + } + }, + ); + it("leaves a write-path failure UNMARKED so the cohort boundary records it once", async () => { // The `upsertPolarMeasurements` transaction throw escapes `syncUserPolar` // uncaught: no source-site record, no mark — the poll-cohort boundary is diff --git a/src/lib/polar/sync.ts b/src/lib/polar/sync.ts index 7878c4475..048445a4f 100644 --- a/src/lib/polar/sync.ts +++ b/src/lib/polar/sync.ts @@ -12,6 +12,22 @@ * as a 401 → `reauth_required` on the shared integration ledger (`polar` key), * exactly like Nightscout's token-rejected path. * + * Isolation: the five vitals collections are settled independently + * (`Promise.allSettled`), so a single dead collection — e.g. a 403 on SpO2 or + * cardio-load on an account without the sensor or the higher tier — imports the + * healthy four and records ONE partial failure rather than blanking every + * collection. Mirrors Oura's shipped per-collection isolation, minus the + * token-refresh leg Polar structurally lacks (a 401 is just a per-collection + * failure the classifier maps to `reauth_required`). + * + * Window contract: every hourly tick re-reads the full server-fixed window for + * each collection — 28 days for recharge / sleep / activity / cardio-load / + * SpO2 (there is no narrower request and no pagination on any of them). A + * late-synced device, a re-scored night, or a HealthLog outage shorter than the + * window all self-heal on the next tick with no cursor or watermark. History + * older than the window is an upstream cap, not a HealthLog gap. See the + * client's per-fetcher docstrings for the verified per-endpoint windows. + * * Idempotency: each row's `externalId` is `:`. The shared * reconciler protects both external and natural identity. Polar re-scores a * night for a short window after the fact, so an exact re-post overwrites in @@ -25,7 +41,6 @@ import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; import { getEvent } from "@/lib/logging/context"; import { - markSyncFailureRecorded, recordSyncFailure, recordSyncSuccess, toFailureKind, @@ -124,61 +139,115 @@ function toUpsert( })); } +/** One Polar vitals collection: its ledger name (which doubles as the + * externalId resource prefix) 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 — a failing fetch OR a throwing mapper — is + * contained to this one collection under `Promise.allSettled`. */ +interface PolarCollectionSpec { + name: string; + collect: (token: string) => Promise; +} + +/** A single collection that failed to fetch/map, carried so the caller can + * classify + record ONE partial failure without blanking the collections that + * did succeed. */ +export interface PolarCollectionFailure { + name: string; + err: unknown; +} + /** * Sync one user's Polar data. Returns the count of measurement rows written. * A user with no Polar connection is a clean no-op (returns 0, touches no * status row). + * + * The five vitals collections are settled independently: the healthy ones + * import, a failing one is recorded as a single partial failure (no success + * stamp), and nothing is rethrown for a fetch/map failure. The only throw that + * escapes now is a write-path/unexpected error out of `upsertPolarMeasurements` + * — the poll-cohort boundary records that unmarked escape once. */ export async function syncUserPolar(userId: string): Promise { const conn = await getPolarConnection(userId); if (!conn) return 0; - const readings: PolarMeasurementUpsert[] = []; + // Filled by the sleep collection's closure as it maps records; carried 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 failed sleep + // collection leaves the sweep list empty and no rows are tombstoned. const sleepSweeps: SleepSegmentSweep[] = []; - try { - const [recharges, sleeps, activities, cardioLoads, spo2Tests] = - await Promise.all([ - fetchNightlyRecharges(conn.accessToken), - fetchSleeps(conn.accessToken), - fetchActivities(conn.accessToken), - fetchCardioLoads(conn.accessToken), - fetchSpo2(conn.accessToken), - ]); - for (const r of recharges) { - readings.push(...toUpsert(mapNightlyRecharge(r), "recharge")); + + const collections: PolarCollectionSpec[] = [ + { + name: "recharge", + collect: async (t) => + (await fetchNightlyRecharges(t)).flatMap((r) => + toUpsert(mapNightlyRecharge(r), "recharge"), + ), + }, + { + name: "sleep", + collect: async (t) => { + const records = await fetchSleeps(t); + const rowsOut: PolarMeasurementUpsert[] = []; + for (const s of records) { + const rows = toUpsert(mapSleep(s), "sleep"); + rowsOut.push(...rows); + // Night-scoped sweep entry: the reconstructed segments of this date + // all key under `sleep::seg:` (mapper-supplied). Any live row + // under that prefix this fetch did NOT re-produce is a re-score + // orphan or a legacy `:seg::` indexed row — tombstoned + // before the upsert. The prefix deliberately stays on `:seg:` — the + // IN_BED envelope keys on its measuredAt's UTC date, which can drift + // a calendar day from `s.date`, so a broader `sleep::` bound + // could cross nights. + sleepSweeps.push({ + prefix: `sleep:${s.date}:seg:`, + keepIds: rows + .filter((r) => r.type === "SLEEP_DURATION") + .map((r) => r.externalId), + }); + } + return rowsOut; + }, + }, + { + name: "activity", + collect: async (t) => + (await fetchActivities(t)).flatMap((a) => + toUpsert(mapActivity(a), "activity"), + ), + }, + { + name: "cardioload", + collect: async (t) => + (await fetchCardioLoads(t)).flatMap((c) => + toUpsert(mapCardioLoad(c), "cardioload"), + ), + }, + { + name: "spo2", + collect: async (t) => + (await fetchSpo2(t)).flatMap((s) => toUpsert(mapSpo2(s), "spo2")), + }, + ]; + + // Isolate each collection: a rejection is captured per-collection, never + // propagated, so one dead collection no longer blanks the healthy siblings. + const settled = await Promise.allSettled( + collections.map((spec) => spec.collect(conn.accessToken)), + ); + const readings: PolarMeasurementUpsert[] = []; + const failures: PolarCollectionFailure[] = []; + settled.forEach((res, i) => { + const spec = collections[i]!; + if (res.status === "fulfilled") { + readings.push(...res.value); + } else { + failures.push({ name: spec.name, err: res.reason }); } - for (const s of sleeps) { - const rows = toUpsert(mapSleep(s), "sleep"); - readings.push(...rows); - // Night-scoped sweep entry: the reconstructed segments of this date all - // key under `sleep::seg:` (mapper-supplied). Any live row under - // that prefix this fetch did NOT re-produce is a re-score orphan or a - // legacy `:seg::` indexed row — tombstoned before the upsert. - // The prefix deliberately stays on `:seg:` — the IN_BED envelope keys - // on its measuredAt's UTC date, which can drift a calendar day from - // `s.date`, so a broader `sleep::` bound could cross nights. - sleepSweeps.push({ - prefix: `sleep:${s.date}:seg:`, - keepIds: rows - .filter((r) => r.type === "SLEEP_DURATION") - .map((r) => r.externalId), - }); - } - for (const a of activities) { - readings.push(...toUpsert(mapActivity(a), "activity")); - } - for (const c of cardioLoads) { - readings.push(...toUpsert(mapCardioLoad(c), "cardioload")); - } - for (const t of spo2Tests) { - readings.push(...toUpsert(mapSpo2(t), "spo2")); - } - } catch (err) { - // Recorded here, then marked so the poll-cohort boundary does not - // double-record it when the legs wrapper rethrows it there. - await recordPolarSyncFailure(userId, err); - throw markSyncFailureRecorded(err); - } + }); // Clear whatever an earlier scoring left under the re-fetched nights before // the fresh set upserts (mirrors Google Health's replace-by-window order). @@ -195,11 +264,42 @@ export async function syncUserPolar(userId: string): Promise { }); // S4 — trigger the debounced morning refresh on a last-night segment landing - // (mirrors the Withings / WHOOP / Apple seams). + // (mirrors the Withings / WHOOP / Apple seams). Fires whether or not a + // sibling collection failed, since the sleep readings were still imported. void maybeEnqueueMorningRefresh(userId, insertedSleepMeasuredAts).catch( () => {}, ); + if (failures.length > 0) { + // Partial failure: import what the healthy collections returned, but keep + // the cycle honest — record ONE 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. Mirrors Oura's + // partial-failure ledger block; the record is inline (not + // `recordPolarSyncFailure`) because it needs the `partial sync failure + // ()` message prefix — the R6 recorder's boundary contract stays + // frozen. No rethrow: a rethrow would suppress the cohort's users_synced / + // reminder-satisfy accounting for a user whose data DID land. + const firstErr = failures[0]!.err; + const names = failures.map((f) => f.name).join(", "); + getEvent()?.addWarning( + `polar: ${failures.length} collection(s) failed for ${userId}: ${names}`, + ); + await recordSyncFailure({ + userId, + integration: "polar", + kind: classifyPolarFailure(firstErr), + message: `partial sync failure (${names}): ${ + firstErr instanceof Error ? firstErr.message : String(firstErr) + }`, + errorCode: + firstErr instanceof PolarApiError && firstErr.httpStatus != null + ? String(firstErr.httpStatus) + : undefined, + }); + return imported; + } + await recordSyncSuccess(userId, "polar"); return imported; }