diff --git a/CHANGELOG.md b/CHANGELOG.md index d9c773c6a..85e4490d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,41 @@ ## [Unreleased] +## [1.32.23] — 2026-07-24 + +Sync status now tells the truth for the connected wearables and CGM sources. A +few paths could report a sync as healthy when it had actually failed, which +reset the failure streak, hid the reconnect prompt, and let a dead grant get +retried on every hourly tick. This release closes those gaps so a failed sync +reads as failed, an interrupted history import is retried instead of marked +complete, and a couple of sources start pulling data sooner. + +- **A dead WHOOP token no longer reports a successful sync.** When the stored + grant is revoked, the connection stays parked with its reconnect prompt and + the last-synced time stays put, instead of flipping back to connected and + hammering the revoked token every hour. This holds on both the manual sync and + the hourly background path. +- **An interrupted WHOOP history import is retried, not marked done.** A + full-history backfill that stops partway is no longer stamped complete, so the + gap is retried rather than frozen in place. To pull deep history after + connecting, run a full sync from Settings. It walks back to 2020. +- **Withings ECG now syncs on its own schedule.** A watch-only account with no + scale previously never had its ECG recordings pulled. An hourly check now + brings them in without waiting on a webhook. +- **Withings activity and sleep write failures now surface.** A failed write no + longer reads as a clean sync; it shows up on the integration's status instead + of leaving a green state over missing data. +- **Oura, Polar, and Nightscout sync errors that previously vanished now + surface.** A write that fails partway through is recorded on the integration + status rather than being silently dropped, and each error is classified and + redacted by the source that owns it. +- **Strava history import starts at connect time.** Deep history begins + importing right after you connect instead of waiting for the next server + restart. +- **WHOOP cycle data uses a wider catch-up window.** A day's strain and energy + figures that are still being revised are re-fetched for a week, matching the + other WHOOP collections, so a late revision is not missed. + ## [1.32.22] — 2026-07-24 The concurrent-edit protection that reached the layout surfaces now covers the diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 9e46f9cf7..b02b00b47 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.32.22 + version: 1.32.23 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index 65445a089..3dbbb9edd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.32.22", + "version": "1.32.23", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index f5da796c6..e9d38b708 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.32.22"; + /* @sw-version-fallback */ "v1.32.23"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/app/api/strava/callback/__tests__/route.test.ts b/src/app/api/strava/callback/__tests__/route.test.ts new file mode 100644 index 000000000..ec238315d --- /dev/null +++ b/src/app/api/strava/callback/__tests__/route.test.ts @@ -0,0 +1,109 @@ +/** + * S-1 — the Strava OAuth callback enqueues the self-converging history backfill + * at connect time (mirroring the WHOOP / Fitbit callbacks) so deep history + * lands within the hour instead of waiting for the next worker reboot. A missing + * boss instance is a warning, never a failure — boot discovery is the net. + */ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { NextRequest } from "next/server"; + +const { + bossSend, + getGlobalBossMock, + exchangeCodeMock, + getCredsMock, + userFindUnique, + userUpdate, + markReconnectedMock, +} = vi.hoisted(() => ({ + bossSend: vi.fn(async () => "job-id"), + getGlobalBossMock: vi.fn(), + exchangeCodeMock: vi.fn(), + getCredsMock: vi.fn(), + userFindUnique: vi.fn(), + userUpdate: vi.fn(async () => ({})), + markReconnectedMock: vi.fn(async () => {}), +})); + +vi.mock("@/lib/api-handler", () => ({ + apiHandler: unknown>(fn: T) => fn, +})); +vi.mock("@/lib/db", () => ({ + prisma: { user: { findUnique: userFindUnique, update: userUpdate } }, +})); +vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn(async () => null) })); +vi.mock("@/lib/logging/context", () => ({ + annotate: vi.fn(), + getEvent: () => ({ + setAuth: vi.fn(), + setError: vi.fn(), + addWarning: vi.fn(), + }), +})); +vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn(async () => {}) })); +vi.mock("@/lib/crypto", () => ({ encrypt: (s: string) => `enc(${s})` })); +vi.mock("@/lib/strava/client", () => ({ exchangeCode: exchangeCodeMock })); +vi.mock("@/lib/strava/credentials", () => ({ + getStravaClientCredentials: getCredsMock, +})); +vi.mock("@/lib/oauth/signed-state", () => ({ + oauthStateCookieName: () => "strava_oauth_state", + stateMatchesCookie: () => true, + verifySignedState: () => ({ userId: "u1" }), +})); +vi.mock("@/lib/integrations/status", () => ({ + markReconnected: markReconnectedMock, +})); +vi.mock("@/lib/jobs/strava-backfill", () => ({ + STRAVA_BACKFILL_QUEUE: "strava-backfill", +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: getGlobalBossMock, +})); + +import { GET } from "../route"; + +process.env.NEXT_PUBLIC_APP_URL = "https://app.example"; + +function callback(): Promise<{ headers: { get(k: string): string | null } }> { + const req = new NextRequest( + "http://localhost/api/strava/callback?code=abc&state=signed-state", + { headers: { cookie: "strava_oauth_state=signed-state" } }, + ); + return (GET as unknown as (r: NextRequest) => Promise)(req); +} + +beforeEach(() => { + vi.clearAllMocks(); + getCredsMock.mockResolvedValue({ clientId: "c", clientSecret: "s" }); + exchangeCodeMock.mockResolvedValue({ + access_token: "at", + refresh_token: "rt", + athlete: { id: 42 }, + }); + userFindUnique.mockResolvedValue({ stravaAthleteId: null }); + getGlobalBossMock.mockReturnValue({ send: bossSend }); +}); + +describe("GET /api/strava/callback — S-1 backfill enqueue", () => { + it("enqueues one STRAVA_BACKFILL_QUEUE job for the connecting user", async () => { + const res = await callback(); + + expect(res.headers.get("location")).toContain("strava=connected"); + expect(markReconnectedMock).toHaveBeenCalledWith("u1", "strava"); + expect(bossSend).toHaveBeenCalledTimes(1); + expect(bossSend).toHaveBeenCalledWith( + "strava-backfill", + expect.objectContaining({ userId: "u1" }), + ); + }); + + it("still connects when the boss instance is unavailable (warning, not failure)", async () => { + getGlobalBossMock.mockReturnValue(null); + + const res = await callback(); + + expect(res.headers.get("location")).toContain("strava=connected"); + expect(bossSend).not.toHaveBeenCalled(); + }); +}); diff --git a/src/app/api/strava/callback/route.ts b/src/app/api/strava/callback/route.ts index dd093e0d6..40934189e 100644 --- a/src/app/api/strava/callback/route.ts +++ b/src/app/api/strava/callback/route.ts @@ -12,6 +12,8 @@ import { verifySignedState, } from "@/lib/oauth/signed-state"; import { markReconnected } from "@/lib/integrations/status"; +import { STRAVA_BACKFILL_QUEUE } from "@/lib/jobs/strava-backfill"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { NextRequest, NextResponse } from "next/server"; /** @@ -107,6 +109,24 @@ export const GET = apiHandler(async (request: NextRequest) => { await auditLog("strava.connect", { userId }); await markReconnected(userId, "strava"); + // Enqueue the self-converging history backfill at connect time so deep + // history lands within the hour instead of waiting for the next worker + // reboot. Best-effort: the boot-time discovery query + // (`stravaBackfillCompletedAt IS NULL`) is the safety net, so a missing + // boss instance here never strands the connection. Mirrors the WHOOP / + // Fitbit callback enqueue. + const boss = getGlobalBoss(); + if (boss) { + await boss + .send(STRAVA_BACKFILL_QUEUE, { + userId, + enqueuedAt: new Date().toISOString(), + }) + .catch((err) => + getEvent()?.addWarning(`strava-backfill enqueue failed: ${err}`), + ); + } + const response = NextResponse.redirect( new URL( "/settings/integrations?strava=connected", diff --git a/src/app/api/whoop/sync/__tests__/route.test.ts b/src/app/api/whoop/sync/__tests__/route.test.ts index 953ae2c02..6e5f461e7 100644 --- a/src/app/api/whoop/sync/__tests__/route.test.ts +++ b/src/app/api/whoop/sync/__tests__/route.test.ts @@ -30,7 +30,7 @@ describe("POST /api/whoop/sync", () => { beforeEach(() => vi.clearAllMocks()); it("triggers an incremental sync by default", async () => { - sync.mockResolvedValue(7); + sync.mockResolvedValue({ imported: 7, failed: false }); const res = (await ( POST as unknown as (r: NextRequest) => Promise<{ data: unknown }> )(req({}))) as { data: { imported: number; fullSync: boolean } }; @@ -40,7 +40,7 @@ describe("POST /api/whoop/sync", () => { }); it("honours fullSync: true", async () => { - sync.mockResolvedValue(99); + sync.mockResolvedValue({ imported: 99, failed: false }); const res = (await ( POST as unknown as (r: NextRequest) => Promise<{ data: unknown }> )(req({ fullSync: true }))) as { data: { fullSync: boolean } }; diff --git a/src/app/api/whoop/sync/route.ts b/src/app/api/whoop/sync/route.ts index 17d069ad4..d3a331aca 100644 --- a/src/app/api/whoop/sync/route.ts +++ b/src/app/api/whoop/sync/route.ts @@ -27,6 +27,6 @@ export const POST = apiHandler(async (request: NextRequest) => { // no body provided -> default incremental sync } - const imported = await syncUserWhoop(user.id, { fullSync }); + const { imported } = await syncUserWhoop(user.id, { fullSync }); return apiSuccess({ imported, fullSync }); }); diff --git a/src/lib/integrations/__tests__/status.test.ts b/src/lib/integrations/__tests__/status.test.ts index 7098e511c..6d78f1f55 100644 --- a/src/lib/integrations/__tests__/status.test.ts +++ b/src/lib/integrations/__tests__/status.test.ts @@ -47,6 +47,8 @@ import { isReauthRequired, getPersistentFailureThreshold, resumeIntegrationFromPark, + markSyncFailureRecorded, + isSyncFailureRecorded, } from "../status"; import { prisma } from "@/lib/db"; import { auditLog } from "@/lib/auth/audit"; @@ -856,3 +858,36 @@ describe("resumeIntegrationFromPark", () => { expect(auditLog).not.toHaveBeenCalled(); }); }); + +describe("sync-failure-recorded marker (§5)", () => { + it("round-trips: a marked error reads back as recorded", () => { + const err = new Error("boom"); + expect(isSyncFailureRecorded(err)).toBe(false); + const returned = markSyncFailureRecorded(err); + // Returns the SAME object so `throw markSyncFailureRecorded(err)` works. + expect(returned).toBe(err); + expect(isSyncFailureRecorded(err)).toBe(true); + }); + + it("the marker is non-enumerable (never serialised into logs / JSON)", () => { + const err = markSyncFailureRecorded(new Error("boom")); + expect(Object.keys(err)).not.toContain("healthlog.syncFailureRecorded"); + expect(JSON.stringify({ ...err })).not.toContain("syncFailureRecorded"); + }); + + it("is a no-op on non-objects and never reports them as recorded", () => { + expect(markSyncFailureRecorded("boom")).toBe("boom"); + expect(markSyncFailureRecorded(null)).toBe(null); + expect(markSyncFailureRecorded(undefined)).toBe(undefined); + expect(isSyncFailureRecorded("boom")).toBe(false); + expect(isSyncFailureRecorded(null)).toBe(false); + expect(isSyncFailureRecorded(undefined)).toBe(false); + }); + + it("tolerates a frozen error without throwing (falls back to unmarked)", () => { + const frozen = Object.freeze(new Error("frozen")); + expect(() => markSyncFailureRecorded(frozen)).not.toThrow(); + // Frozen can't carry the marker, so the boundary records it once — safe. + expect(isSyncFailureRecorded(frozen)).toBe(false); + }); +}); diff --git a/src/lib/integrations/status.ts b/src/lib/integrations/status.ts index 953e0f4b9..8e87682b6 100644 --- a/src/lib/integrations/status.ts +++ b/src/lib/integrations/status.ts @@ -98,6 +98,51 @@ export function toFailureKind( return "transient"; } +/** + * Non-enumerable marker stamped on an error a provider sync ALREADY recorded on + * the integration ledger before rethrowing. The poll-cohort boundary + * (`makePollCohortHandler`) reads it so a record-then-rethrow failure is not + * recorded a SECOND time when it surfaces in the cohort catch — a double-record + * would inflate the per-kind failure buckets (firing the 3-strike admin alert a + * tick early) and churn `lastError`. A provider-blind recorder at the boundary + * would also persist an UNREDACTED message (e.g. Nightscout's token-bearing + * URL), so the marker keeps classification + redaction provider-owned: only an + * UNMARKED escape reaches the boundary recorder. + * + * No-op on non-objects; tolerant of a frozen error (the boundary then records + * it once, which is the correct fallback — never a double-record). + */ +const SYNC_FAILURE_RECORDED: unique symbol = Symbol( + "healthlog.syncFailureRecorded", +); + +/** Stamp `err` as already-recorded and return it (for `throw markSync…(err)`). */ +export function markSyncFailureRecorded(err: T): T { + if (err !== null && typeof err === "object") { + try { + Object.defineProperty(err, SYNC_FAILURE_RECORDED, { + value: true, + enumerable: false, + configurable: true, + writable: true, + }); + } catch { + // A frozen / sealed error can't carry the marker; the boundary records + // it once — the safe fallback, never a double-record. + } + } + return err; +} + +/** True when `err` was already recorded on the ledger by its source site. */ +export function isSyncFailureRecorded(err: unknown): boolean { + return ( + err !== null && + typeof err === "object" && + (err as Record)[SYNC_FAILURE_RECORDED] === true + ); +} + /** * Recognised IntegrationStatus states. * diff --git a/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts b/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts index 710f923b9..7b816dc90 100644 --- a/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts +++ b/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts @@ -107,7 +107,7 @@ describe("enqueueBootTimeSleepTimelineBackfill — discovery", () => { describe("runSleepTimelineBackfillForUser", () => { it("WHOOP: deletes the source's sleep rows, full-syncs, and stamps the marker", async () => { prismaMock.measurement.deleteMany.mockResolvedValue({ count: 5 }); - syncUserWhoop.mockResolvedValue(42); + syncUserWhoop.mockResolvedValue({ imported: 42, failed: false }); prismaMock.whoopConnection.update.mockResolvedValue({}); const { deleted, imported } = await runSleepTimelineBackfillForUser( @@ -126,6 +126,18 @@ describe("runSleepTimelineBackfillForUser", () => { expect(updateArg.data.sleepTimelineBackfillAt).toBeInstanceOf(Date); }); + it("WHOOP: a failed verdict THROWS without stamping the marker (gated re-sync)", async () => { + // The rows were already deleted; a partial re-sync must not stamp the + // completion marker over the gap — it throws so pg-boss retries. + prismaMock.measurement.deleteMany.mockResolvedValue({ count: 5 }); + syncUserWhoop.mockResolvedValue({ imported: 3, failed: true }); + + await expect( + runSleepTimelineBackfillForUser("w1", "WHOOP"), + ).rejects.toThrow(/incomplete/); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + }); + it("WITHINGS: deletes the source's sleep rows, re-syncs, and stamps the marker", async () => { prismaMock.measurement.deleteMany.mockResolvedValue({ count: 3 }); syncWithingsSleep.mockResolvedValue(7); diff --git a/src/lib/jobs/__tests__/whoop-backfill.test.ts b/src/lib/jobs/__tests__/whoop-backfill.test.ts index e6573ad2d..7cb6404e4 100644 --- a/src/lib/jobs/__tests__/whoop-backfill.test.ts +++ b/src/lib/jobs/__tests__/whoop-backfill.test.ts @@ -7,16 +7,18 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { prismaMock, bossSend, syncUserWhoop } = vi.hoisted(() => ({ - prismaMock: { - whoopConnection: { - findMany: vi.fn(), - update: vi.fn(), +const { prismaMock, bossSend, syncUserWhoop, isReauthRequiredMock } = + vi.hoisted(() => ({ + prismaMock: { + whoopConnection: { + findMany: vi.fn(), + update: vi.fn(), + }, }, - }, - bossSend: vi.fn(), - syncUserWhoop: vi.fn(), -})); + bossSend: vi.fn(), + syncUserWhoop: vi.fn(), + isReauthRequiredMock: vi.fn(async () => false), + })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); @@ -28,6 +30,10 @@ vi.mock("@/lib/whoop/sync", () => ({ syncUserWhoop: (...a: unknown[]) => syncUserWhoop(...a), })); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: isReauthRequiredMock, +})); + vi.mock("@/lib/logging/context", () => ({ annotate: () => {}, getEvent: () => null, @@ -40,6 +46,8 @@ import { beforeEach(() => { vi.clearAllMocks(); + isReauthRequiredMock.mockResolvedValue(false); + prismaMock.whoopConnection.update.mockResolvedValue({}); }); describe("enqueueBootTimeWhoopBackfill — discovery", () => { @@ -85,10 +93,9 @@ describe("enqueueBootTimeWhoopBackfill — discovery", () => { }); }); -describe("runWhoopBackfillForUser", () => { - it("runs a full sync and stamps backfillCompletedAt", async () => { - syncUserWhoop.mockResolvedValue(123); - prismaMock.whoopConnection.update.mockResolvedValue({}); +describe("runWhoopBackfillForUser — verdict-gated marker", () => { + it("stamps backfillCompletedAt after a CLEAN full-history run", async () => { + syncUserWhoop.mockResolvedValue({ imported: 123, failed: false }); const { imported } = await runWhoopBackfillForUser("u1"); @@ -98,4 +105,21 @@ describe("runWhoopBackfillForUser", () => { expect(updateArg.where).toEqual({ userId: "u1" }); expect(updateArg.data.backfillCompletedAt).toBeInstanceOf(Date); }); + + it("a failed verdict THROWS without stamping — pg-boss retries become real", async () => { + syncUserWhoop.mockResolvedValue({ imported: 7, failed: true }); + + await expect(runWhoopBackfillForUser("u1")).rejects.toThrow(/incomplete/); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + }); + + it("a connection parked at error_reauth returns WITHOUT running the sync or stamping", async () => { + isReauthRequiredMock.mockResolvedValue(true); + + await expect(runWhoopBackfillForUser("u1")).resolves.toEqual({ + imported: 0, + }); + expect(syncUserWhoop).not.toHaveBeenCalled(); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + }); }); diff --git a/src/lib/jobs/__tests__/whoop-resource-sync.test.ts b/src/lib/jobs/__tests__/whoop-resource-sync.test.ts index a7469d718..295ae98e6 100644 --- a/src/lib/jobs/__tests__/whoop-resource-sync.test.ts +++ b/src/lib/jobs/__tests__/whoop-resource-sync.test.ts @@ -10,17 +10,38 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { Job } from "pg-boss"; -const { syncUserWorkout, syncWhoopWorkoutById, findManyMock } = vi.hoisted( - () => ({ - syncUserWorkout: vi.fn<(...a: unknown[]) => Promise>(async () => 0), - syncWhoopWorkoutById: vi.fn<(...a: unknown[]) => Promise>( - async () => 0, - ), - findManyMock: vi.fn< - (...a: unknown[]) => Promise> - >(async () => []), - }), -); +const { + syncUserWorkout, + syncWhoopWorkoutById, + findManyMock, + recordWhoopSyncFailureMock, + recordSyncSuccessMock, +} = vi.hoisted(() => ({ + syncUserWorkout: vi.fn<(...a: unknown[]) => Promise>(async () => 0), + syncWhoopWorkoutById: vi.fn<(...a: unknown[]) => Promise>( + async () => 0, + ), + findManyMock: vi.fn<(...a: unknown[]) => Promise>>( + async () => [], + ), + recordWhoopSyncFailureMock: vi.fn<(...a: unknown[]) => Promise>( + async () => {}, + ), + recordSyncSuccessMock: vi.fn<(...a: unknown[]) => Promise>( + async () => {}, + ), +})); + +// Keep the real marker helpers (isSyncFailureRecorded / markSyncFailureRecorded) +// but stub the DB-touching writers so this stays a unit test. +vi.mock("@/lib/integrations/status", async (orig) => ({ + ...(await orig()), + recordSyncSuccess: recordSyncSuccessMock, +})); +vi.mock("@/lib/whoop/sync-core", async (orig) => ({ + ...(await orig()), + recordWhoopSyncFailure: recordWhoopSyncFailureMock, +})); vi.mock("@/lib/whoop/sync-recovery", () => ({ syncUserRecovery: vi.fn(async () => 0), @@ -63,6 +84,7 @@ vi.mock("../reminder/shared", () => ({ })); import { handleWhoopWorkoutSync } from "../reminder/whoop-sync"; +import { markSyncFailureRecorded } from "@/lib/integrations/status"; function job(data: { userId?: string; resourceId?: string }): Job<{ userId?: string; @@ -111,3 +133,32 @@ describe("runWhoopResourceSync dispatch", () => { expect(syncUserWorkout).toHaveBeenCalledTimes(2); }); }); + +describe("runWhoopResourceSync — WH-4 write-path ledger honesty", () => { + it("records an UNMARKED write-path throw once (a stale-green pill otherwise)", async () => { + // A `MeasurementReconciliationError` out of the upsert reaches no ledger on + // the cron path before WH-4 — the pill stays green through a persistent + // write failure. The catch now records it once. + syncUserWorkout.mockRejectedValueOnce(new Error("write failed")); + + await handleWhoopWorkoutSync([job({ userId: "u1" })]); + + expect(recordWhoopSyncFailureMock).toHaveBeenCalledTimes(1); + expect(recordWhoopSyncFailureMock).toHaveBeenCalledWith( + "u1", + expect.objectContaining({ message: "write failed" }), + ); + }); + + it("does NOT re-record a MARKED fetch error (no double-record)", async () => { + // A fetch hard-fail was already recorded + marked in + // `handleCollectionFetchError`; the cron catch must skip it. + syncUserWorkout.mockRejectedValueOnce( + markSyncFailureRecorded(new Error("already recorded fetch error")), + ); + + await handleWhoopWorkoutSync([job({ userId: "u1" })]); + + expect(recordWhoopSyncFailureMock).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/jobs/__tests__/withings-ecg-worker.test.ts b/src/lib/jobs/__tests__/withings-ecg-worker.test.ts index 05de54a0a..cb11bde9e 100644 --- a/src/lib/jobs/__tests__/withings-ecg-worker.test.ts +++ b/src/lib/jobs/__tests__/withings-ecg-worker.test.ts @@ -52,9 +52,15 @@ vi.mock("@/lib/logging/background", () => ({ vi.mock("@/lib/logging/fire-and-forget", () => ({ fireAndForget: vi.fn(), })); +const { ecgFindMany } = vi.hoisted(() => ({ + ecgFindMany: vi.fn<(...a: unknown[]) => Promise>>( + async () => [], + ), +})); + vi.mock("../reminder/shared", () => ({ getWorkerPrisma: vi.fn(() => ({ - withingsConnection: { findMany: vi.fn(async () => []) }, + withingsConnection: { findMany: (...a: unknown[]) => ecgFindMany(...a) }, })), })); @@ -69,6 +75,7 @@ function job(data: WithingsEcgSyncPayload): Job { beforeEach(() => { vi.clearAllMocks(); + ecgFindMany.mockResolvedValue([]); }); const registrarSource = readFileSync( join(process.cwd(), "src/lib/jobs/reminder/register-integration-sync.ts"), @@ -76,10 +83,8 @@ const registrarSource = readFileSync( ); describe("handleWithingsEcgSync", () => { - it("rejects a failed source write so pg-boss can retry the same job", async () => { - syncUserEcg - .mockRejectedValueOnce(new Error("source write failed")) - .mockResolvedValueOnce(1); + it("a webhook job (userId + window) does a targeted date-windowed sync", async () => { + syncUserEcg.mockResolvedValue(1); const queued = job({ userId: "user-1", eventId: "wu-1:1:1715000000:1715000060", @@ -88,20 +93,42 @@ describe("handleWithingsEcgSync", () => { enddate: 1715000060, }); - await expect(handleWithingsEcgSync([queued])).rejects.toThrow( - "source write failed", - ); await expect(handleWithingsEcgSync([queued])).resolves.toBeUndefined(); + expect(ecgFindMany).not.toHaveBeenCalled(); + expect(syncUserEcg).toHaveBeenCalledTimes(1); + expect(syncUserEcg).toHaveBeenCalledWith("user-1", { + startdate: 1715000000, + enddate: 1715000060, + }); + }); + + it("a payload-less cron tick walks the connection cohort with the default window", async () => { + // W-1 — the catch-net that lets a watch-only account (no scale, appli 54 + // unsubscribed) pull its ECG strips: no userId → iterate every connection + // with `syncUserEcg(userId)` (default trailing-30-day window). + ecgFindMany.mockResolvedValue([{ userId: "a" }, { userId: "b" }]); + syncUserEcg.mockResolvedValue(0); + + await handleWithingsEcgSync([job({} as WithingsEcgSyncPayload)]); + + expect(ecgFindMany).toHaveBeenCalledTimes(1); + expect(syncUserEcg).toHaveBeenCalledTimes(2); + expect(syncUserEcg).toHaveBeenNthCalledWith(1, "a", {}); + expect(syncUserEcg).toHaveBeenNthCalledWith(2, "b", {}); + }); + + it("one user's failure never starves the rest of the cohort", async () => { + ecgFindMany.mockResolvedValue([{ userId: "a" }, { userId: "b" }]); + syncUserEcg.mockRejectedValueOnce(new Error("boom for a")); + + // Per-user catch-and-warn (mirrors the activity / sleep handlers): the + // handler resolves, both users attempted, the leaf records its own ledger + // failure so honesty is preserved without rejecting the whole batch. + await expect( + handleWithingsEcgSync([job({} as WithingsEcgSyncPayload)]), + ).resolves.toBeUndefined(); expect(syncUserEcg).toHaveBeenCalledTimes(2); - expect(syncUserEcg).toHaveBeenNthCalledWith(1, "user-1", { - startdate: 1715000000, - enddate: 1715000060, - }); - expect(syncUserEcg).toHaveBeenNthCalledWith(2, "user-1", { - startdate: 1715000000, - enddate: 1715000060, - }); }); }); diff --git a/src/lib/jobs/__tests__/withings-queues.test.ts b/src/lib/jobs/__tests__/withings-queues.test.ts index c525e1a04..5dc7dac08 100644 --- a/src/lib/jobs/__tests__/withings-queues.test.ts +++ b/src/lib/jobs/__tests__/withings-queues.test.ts @@ -76,4 +76,15 @@ describe("reminder-worker — Withings activity + sleep v2 queues", () => { ); expect(source).toMatch(/\[WITHINGS_SLEEP_QUEUE,\s*WITHINGS_SLEEP_CRON\]/); }); + + it("schedules the ECG catch-net cron at :41 so watch-only accounts pull strips", () => { + // The queue is already in allQueues + boss.work-bound; W-1 adds the missing + // schedule entry so a payload-less cron tick walks the connection cohort. + expect(source).toMatch( + /WITHINGS_ECG_SYNC_CRON\s*=\s*["']41 \* \* \* \*["']/, + ); + expect(source).toMatch( + /\[WITHINGS_ECG_SYNC_QUEUE,\s*WITHINGS_ECG_SYNC_CRON\]/, + ); + }); }); 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 65ab44cd4..3d1c30923 100644 --- a/src/lib/jobs/reminder/__tests__/polar-sync-legs.test.ts +++ b/src/lib/jobs/reminder/__tests__/polar-sync-legs.test.ts @@ -5,7 +5,12 @@ const { syncUserPolar, syncUserPolarWorkouts } = vi.hoisted(() => ({ syncUserPolarWorkouts: vi.fn(), })); -vi.mock("@/lib/polar/sync", () => ({ syncUserPolar })); +vi.mock("@/lib/polar/sync", () => ({ + syncUserPolar, + // The cohort config wires `recordPolarSyncFailure` as its recorder; the legs + // handler imports the module, so the mock must expose it too. + recordPolarSyncFailure: vi.fn(async () => {}), +})); vi.mock("@/lib/polar/sync-workouts", () => ({ syncUserPolarWorkouts })); import { syncUserPolarLegs } from "../polar-sync"; diff --git a/src/lib/jobs/reminder/__tests__/poll-cohort.test.ts b/src/lib/jobs/reminder/__tests__/poll-cohort.test.ts new file mode 100644 index 000000000..f0dc9123d --- /dev/null +++ b/src/lib/jobs/reminder/__tests__/poll-cohort.test.ts @@ -0,0 +1,115 @@ +/** + * §5 — the poll-cohort factory boundary. A provider that records-then-rethrows + * marks its error so the boundary does NOT record it a second time (which would + * inflate the per-kind buckets and, for Nightscout, risk re-persisting a + * token-bearing URL). An UNMARKED escape — today's Oura / Polar write-path throw + * — is recorded exactly once through the per-provider `recordFailure` callback. + * A recorder rejection never breaks the cohort loop. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job } from "pg-boss"; + +const warnings: string[] = []; + +vi.mock("@/lib/jobs/worker-status", () => ({ recordError: vi.fn() })); +vi.mock("@/lib/jobs/reminder-satisfy", () => ({ + enqueueReminderSatisfy: vi.fn(async () => undefined), +})); +vi.mock("@/lib/logging/fire-and-forget", () => ({ fireAndForget: vi.fn() })); +vi.mock("@/lib/logging/background", () => ({ + withBackgroundEvent: async ( + _name: string, + fn: (evt: { + setBackground: () => void; + setError: () => void; + addWarning: (w: string) => void; + }) => Promise, + ) => + fn({ + setBackground: vi.fn(), + setError: vi.fn(), + addWarning: (w: string) => warnings.push(w), + }), +})); +vi.mock("../shared", () => ({ + getWorkerPrisma: () => ({}), +})); + +import { makePollCohortHandler } from "../poll-cohort"; +import { markSyncFailureRecorded } from "@/lib/integrations/status"; + +function job(userId?: string): Job<{ userId?: string }> { + return { data: userId ? { userId } : {} } as Job<{ userId?: string }>; +} + +beforeEach(() => { + warnings.length = 0; + vi.clearAllMocks(); +}); + +describe("makePollCohortHandler — marker-aware recordFailure boundary", () => { + it("records an UNMARKED escape once, then warns, and continues the cohort", async () => { + const recordFailure = vi.fn(async () => {}); + const syncUser = vi.fn(async (userId: string) => { + if (userId === "boom") throw new Error("unmarked write failure"); + return 1; + }); + const handler = makePollCohortHandler({ + taskName: "job.test_sync", + findCohort: async () => ["a", "boom", "b"], + syncUser, + recordFailure, + }); + + await handler([job()]); + + // The failing user is recorded once and warned; the other two still sync. + expect(syncUser).toHaveBeenCalledTimes(3); + expect(recordFailure).toHaveBeenCalledTimes(1); + expect(recordFailure).toHaveBeenCalledWith( + "boom", + expect.objectContaining({ message: "unmarked write failure" }), + ); + expect(warnings.some((w) => w.includes("boom"))).toBe(true); + }); + + it("does NOT record a MARKED error (the source already recorded it)", async () => { + const recordFailure = vi.fn(async () => {}); + const syncUser = vi.fn(async () => { + throw markSyncFailureRecorded(new Error("already recorded at source")); + }); + const handler = makePollCohortHandler({ + taskName: "job.test_sync", + findCohort: async () => ["a"], + syncUser, + recordFailure, + }); + + await handler([job()]); + + expect(recordFailure).not.toHaveBeenCalled(); + // Still warned so the cohort event carries the failure. + expect(warnings.some((w) => w.includes("already recorded"))).toBe(true); + }); + + it("a recorder rejection never breaks the cohort loop", async () => { + const recordFailure = vi.fn(async () => { + throw new Error("recorder down"); + }); + const syncUser = vi.fn(async (userId: string) => { + if (userId === "boom") throw new Error("unmarked"); + return 1; + }); + const handler = makePollCohortHandler({ + taskName: "job.test_sync", + findCohort: async () => ["boom", "next"], + syncUser, + recordFailure, + }); + + await expect(handler([job()])).resolves.toBeUndefined(); + // The next user still synced despite the recorder throwing. + expect(syncUser).toHaveBeenCalledTimes(2); + expect(syncUser).toHaveBeenLastCalledWith("next"); + }); +}); diff --git a/src/lib/jobs/reminder/nightscout-sync.ts b/src/lib/jobs/reminder/nightscout-sync.ts index 1a97f24b1..76cbc7b57 100644 --- a/src/lib/jobs/reminder/nightscout-sync.ts +++ b/src/lib/jobs/reminder/nightscout-sync.ts @@ -9,7 +9,10 @@ * `./poll-cohort`; reminder-worker.ts owns the queue name, cron schedule, and * boss.work registration. */ -import { syncUserNightscout } from "@/lib/nightscout/sync"; +import { + recordNightscoutSyncFailure, + syncUserNightscout, +} from "@/lib/nightscout/sync"; import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort"; export type NightscoutSyncPayload = PollCohortPayload; @@ -25,4 +28,8 @@ export const handleNightscoutSync = makePollCohortHandler({ return users.map((u) => u.id); }, syncUser: syncUserNightscout, + // The fetch path already records + marks its own failure (token-redacted), so + // this fires only for a future UNMARKED escape — and redacts the token before + // the message reaches the ledger. + recordFailure: recordNightscoutSyncFailure, }); diff --git a/src/lib/jobs/reminder/oura-sync.ts b/src/lib/jobs/reminder/oura-sync.ts index 5ca264c4e..a22d2f9ad 100644 --- a/src/lib/jobs/reminder/oura-sync.ts +++ b/src/lib/jobs/reminder/oura-sync.ts @@ -7,7 +7,7 @@ * lives in `./poll-cohort`; the queue name, cron schedule, and boss.work * registration live in reminder-worker.ts. */ -import { syncUserOura } from "@/lib/oura/sync"; +import { recordOuraSyncFailure, syncUserOura } from "@/lib/oura/sync"; import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort"; export type OuraSyncPayload = PollCohortPayload; @@ -22,4 +22,7 @@ export const handleOuraSync = makePollCohortHandler({ return users.map((u) => u.id); }, syncUser: syncUserOura, + // Catches an UNMARKED escape — today the `upsertOuraMeasurements` write-path + // throw (the fetch/refresh path already records + marks its own failures). + recordFailure: recordOuraSyncFailure, }); diff --git a/src/lib/jobs/reminder/polar-sync.ts b/src/lib/jobs/reminder/polar-sync.ts index b73efee48..ae06b02c9 100644 --- a/src/lib/jobs/reminder/polar-sync.ts +++ b/src/lib/jobs/reminder/polar-sync.ts @@ -13,7 +13,7 @@ * reminder-worker.ts — no new queue: the workout leg rides the existing hourly * `polar-sync` tick (one extra request per connected user per hour). */ -import { syncUserPolar } from "@/lib/polar/sync"; +import { recordPolarSyncFailure, syncUserPolar } from "@/lib/polar/sync"; import { syncUserPolarWorkouts } from "@/lib/polar/sync-workouts"; import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort"; @@ -63,4 +63,8 @@ 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). + recordFailure: recordPolarSyncFailure, }); diff --git a/src/lib/jobs/reminder/poll-cohort.ts b/src/lib/jobs/reminder/poll-cohort.ts index 2e716552b..a24a12ff3 100644 --- a/src/lib/jobs/reminder/poll-cohort.ts +++ b/src/lib/jobs/reminder/poll-cohort.ts @@ -23,6 +23,7 @@ import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError } from "@/lib/jobs/worker-status"; import { withBackgroundEvent } from "@/lib/logging/background"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; +import { isSyncFailureRecorded } from "@/lib/integrations/status"; import { getWorkerPrisma } from "./shared"; /** Every poll-cohort handler accepts an optional single-user payload. */ @@ -40,6 +41,16 @@ export interface PollCohortConfig { findCohort: (prisma: ReturnType) => Promise; /** Sync one user; returns the count of measurements imported. */ syncUser: (userId: string) => Promise; + /** + * Record a per-user sync failure that escaped `syncUser` UNRECORDED. Called + * only for an error the source site did not already mark via + * `markSyncFailureRecorded` — so a provider that records-then-rethrows never + * double-records, and the recorder keeps classification + secret redaction + * provider-owned (a boundary-blind recorder would leak Nightscout's + * token-bearing URL into `lastError`). A rejection here never breaks the + * cohort loop. + */ + recordFailure: (userId: string, err: unknown) => Promise; } /** @@ -52,6 +63,7 @@ export function makePollCohortHandler({ taskName, findCohort, syncUser, + recordFailure, }: PollCohortConfig) { return async function handlePollCohort(jobs: Job[]) { await withBackgroundEvent(taskName, async (evt) => { @@ -82,6 +94,18 @@ export function makePollCohortHandler({ }); } } catch (err) { + // A record-then-rethrow provider already stamped the ledger and + // marked the error; recording again would double-count the streak + // and (for Nightscout) risk re-persisting an unredacted message. + // Only an UNMARKED escape — today's Oura / Polar write-path throw, + // or any future provider path — is recorded here. + if (!isSyncFailureRecorded(err)) { + await recordFailure(userId, err).catch((recErr) => { + evt.addWarning( + `${taskName} recordFailure failed for user ${userId}: ${recErr}`, + ); + }); + } evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } } diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index 6fe050510..eaaf55544 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -137,6 +137,14 @@ const WITHINGS_ACTIVITY_CRON = "0 * * * *"; // every hour at :00 const WITHINGS_SLEEP_QUEUE = "withings-sleep-sync"; const WITHINGS_SLEEP_CRON = "15 * * * *"; // every hour at :15 +// v1.32.23 — Withings ECG cron catch-net. The queue is webhook-primary +// (appli=54 → `enqueueWithingsEcgSync`), but appli 54 is not in the +// subscription set and the manual sync route measures only, so a ScanWatch-only +// account (no scale) never had its ECG strips pulled. An hourly cohort walk +// (one `v2/heart list` call per connected user, empty + cheap for non-ECG +// devices) closes the gap. :41 is a free minute (used: 0,5,8,11,13,15,17,18, +// 20,30,35,50) staggered off every other hourly sync tick. +const WITHINGS_ECG_SYNC_CRON = "41 * * * *"; // every hour at :41 const MOODLOG_SYNC_QUEUE = "moodlog-sync"; @@ -310,6 +318,9 @@ const schedules: ScheduleEntry[] = [ [WITHINGS_SYNC_QUEUE, WITHINGS_SYNC_CRON], [WITHINGS_ACTIVITY_QUEUE, WITHINGS_ACTIVITY_CRON], [WITHINGS_SLEEP_QUEUE, WITHINGS_SLEEP_CRON], + // v1.32.23 — hourly ECG catch-net (:41) so watch-only accounts pull their + // strips without a webhook or a scale. + [WITHINGS_ECG_SYNC_QUEUE, WITHINGS_ECG_SYNC_CRON], [MOODLOG_SYNC_QUEUE, MOODLOG_SYNC_CRON], [WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, WITHINGS_OAUTH_STATE_CLEANUP_CRON], // v1.11.0 — WHOOP poll-fallback crons. Recovery/sleep/workout catch diff --git a/src/lib/jobs/reminder/strava-sync.ts b/src/lib/jobs/reminder/strava-sync.ts index b3e6e1b0d..660d2a9d8 100644 --- a/src/lib/jobs/reminder/strava-sync.ts +++ b/src/lib/jobs/reminder/strava-sync.ts @@ -11,7 +11,7 @@ * The queue name, cron schedule, and boss.work registration live in * `register-integration-sync.ts`. */ -import { syncUserStrava } from "@/lib/strava/sync"; +import { recordStravaFailure, syncUserStrava } from "@/lib/strava/sync"; import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort"; export type StravaSyncPayload = PollCohortPayload; @@ -26,4 +26,8 @@ export const handleStravaSync = makePollCohortHandler({ return users.map((u) => u.id); }, syncUser: (userId) => syncUserStrava(userId), + // Strava records + marks every failure at its source today, so this is inert + // now — it's the net for a future unmarked escape (matching the cohort's + // contract that every provider supplies a recorder). + recordFailure: recordStravaFailure, }); diff --git a/src/lib/jobs/reminder/whoop-sync.ts b/src/lib/jobs/reminder/whoop-sync.ts index a99c33358..b8df364f5 100644 --- a/src/lib/jobs/reminder/whoop-sync.ts +++ b/src/lib/jobs/reminder/whoop-sync.ts @@ -21,7 +21,11 @@ import { syncUserWorkout, syncWhoopWorkoutById, } from "@/lib/whoop/sync-workout"; -import { syncWhoopResourceWithStatus } from "@/lib/whoop/sync-core"; +import { + recordWhoopSyncFailure, + syncWhoopResourceWithStatus, +} from "@/lib/whoop/sync-core"; +import { isSyncFailureRecorded } from "@/lib/integrations/status"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { getWorkerPrisma } from "./shared"; @@ -110,6 +114,13 @@ export async function runWhoopResourceSync( }); } } catch (err) { + // WH-4: a write-path throw on the hourly cron reaches no ledger + // otherwise (stale-green pill). Fetch hard-fails already recorded + + // marked in `handleCollectionFetchError`, so record only the UNMARKED + // escapes here. + if (!isSyncFailureRecorded(err)) { + await recordWhoopSyncFailure(userId, err); + } evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } } diff --git a/src/lib/jobs/reminder/withings-sync.ts b/src/lib/jobs/reminder/withings-sync.ts index fbfd828a8..96be0b354 100644 --- a/src/lib/jobs/reminder/withings-sync.ts +++ b/src/lib/jobs/reminder/withings-sync.ts @@ -108,25 +108,72 @@ export async function handleWithingsFallbackSync( } /** - * Drain webhook-owned ECG events. A failed source read or write is allowed to - * reject the handler so pg-boss retains the job and applies its retry policy. + * ECG sync handler. Two enqueue paths feed this queue, mirroring the + * activity / sleep handlers: + * + * 1. Webhook (appli=54) — payload carries `userId` (+ the source + * `startdate`/`enddate` window); the handler runs `syncUserEcg` for that + * one user over the windowed range. + * 2. Cron (`withings-ecg-sync` at :41) — payload has no `userId`, the handler + * iterates every Withings connection and runs `syncUserEcg` over its + * default trailing 30-day window. This is the catch-net that lets a + * watch-only account (no scale, appli 54 unsubscribed) pull its ECG strips + * without a webhook. + * + * Per-user failures are warned and the cohort carries on so one user's + * parked-at-reauth state (or a transient device outage) never starves the rest; + * the ECG leaf records its own classified failure on the shared `withings` + * ledger, so honesty is preserved without rejecting the whole batch. */ export async function handleWithingsEcgSync( jobs: Job[], ) { await withBackgroundEvent("job.withings_ecg_sync", async (evt) => { + const prisma = getWorkerPrisma(); try { - let recordingsImported = 0; + const targets: Array<{ + userId: string; + startdate?: number; + enddate?: number; + }> = []; for (const job of jobs) { - recordingsImported += await syncUserEcg(job.data.userId, { - startdate: job.data.startdate, - enddate: job.data.enddate, - }); + if (job.data?.userId) { + targets.push({ + userId: job.data.userId, + startdate: job.data.startdate, + enddate: job.data.enddate, + }); + } } + // No user-specific enqueue → cron fallback iterating every connection. + if (targets.length === 0) { + const connections = await prisma.withingsConnection.findMany({ + select: { userId: true }, + }); + targets.push(...connections); + } + if (targets.length === 0) return; + + let usersSynced = 0; + let recordingsImported = 0; + for (const { userId, startdate, enddate } of targets) { + try { + const options = + startdate === undefined && enddate === undefined + ? {} + : { startdate, enddate }; + recordingsImported += await syncUserEcg(userId, options); + usersSynced++; + } catch (err) { + evt.addWarning(`Withings ECG sync failed for user ${userId}: ${err}`); + } + } + evt.setBackground({ task_name: "job.withings_ecg_sync", result: { - events_processed: jobs.length, + users_synced: usersSynced, + total: targets.length, recordings_imported: recordingsImported, }, }); diff --git a/src/lib/jobs/sleep-timeline-backfill.ts b/src/lib/jobs/sleep-timeline-backfill.ts index 5cee00849..33aed68bf 100644 --- a/src/lib/jobs/sleep-timeline-backfill.ts +++ b/src/lib/jobs/sleep-timeline-backfill.ts @@ -71,7 +71,17 @@ export async function runSleepTimelineBackfillForUser( let imported = 0; if (provider === "WHOOP") { - imported = await syncUserWhoop(userId, { fullSync: true }); + // `syncUserWhoop` now returns a verdict; gate the completion marker on a + // clean run so a partial re-sync (dead token, write failure) throws for a + // pg-boss retry instead of freezing the timeline gap under a stamped + // marker — the same discipline the WHOOP/Fitbit backfills use. + const result = await syncUserWhoop(userId, { fullSync: true }); + imported = result.imported; + if (result.failed) { + throw new Error( + `sleep-timeline backfill incomplete for user ${userId} (whoop) — marker not stamped`, + ); + } await prisma.whoopConnection.update({ where: { userId }, data: { sleepTimelineBackfillAt: new Date() }, diff --git a/src/lib/jobs/whoop-backfill.ts b/src/lib/jobs/whoop-backfill.ts index aded98c8e..488ae6b1c 100644 --- a/src/lib/jobs/whoop-backfill.ts +++ b/src/lib/jobs/whoop-backfill.ts @@ -20,6 +20,7 @@ import { prisma } from "@/lib/db"; import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; +import { isReauthRequired } from "@/lib/integrations/status"; import { syncUserWhoop } from "@/lib/whoop/sync"; export const WHOOP_BACKFILL_QUEUE = "whoop-backfill"; @@ -37,15 +38,46 @@ export interface WhoopBackfillPayload { } /** - * Per-user backfill handler. Runs a full-history sync for one account and - * stamps `backfillCompletedAt` so the discovery query drops it. Idempotent: - * the per-resource upserts are key-stable, so a re-run (e.g. a reboot mid-walk) - * overwrites rather than duplicating. + * Per-user backfill handler. Runs a full-history sync for one account and stamps + * `backfillCompletedAt` ONLY on a clean run, so the discovery query drops it. + * Idempotent: the per-resource upserts are key-stable, so a re-run (e.g. a + * reboot mid-walk) overwrites rather than duplicating. Mirrors the landed + * `runFitbitBackfillForUser`. + * + * Verdict-gated: a partial hard failure (a dead token, a write failing, a + * fetch hard-fail) THROWS so pg-boss retries the job — stamping the marker over + * a partial multi-year walk would freeze the history gap forever and the + * configured `retryLimit` would be dead code. A connection parked at + * `error_reauth` is NOT clean either, but throwing against a dead grant would + * just burn the retry budget on the same no-op: return WITHOUT stamping — boot + * discovery re-offers the account after the user reconnects and a clean walk + * completes. `syncUserWhoop` has its own reauth short-circuit (returns + * `failed: true`), but the guard must still live here so a parked account + * RETURNS instead of throwing through three retries. */ export async function runWhoopBackfillForUser( userId: string, ): Promise<{ imported: number }> { - const imported = await syncUserWhoop(userId, { fullSync: true }); + if (await isReauthRequired(userId, "whoop")) { + annotate({ + action: { + name: "whoop.backfill.skipped_reauth", + details: { imported: 0 }, + }, + }); + return { imported: 0 }; + } + + const { imported, failed } = await syncUserWhoop(userId, { fullSync: true }); + + if (failed) { + // Surface through the pg-boss retry path (retryLimit + backoff set by + // `integrationBackfillSourceOptions`). If the failure parked the connection + // at error_reauth meanwhile, the next attempt takes the reauth return above. + throw new Error( + `whoop backfill incomplete for user ${userId} — marker not stamped`, + ); + } await prisma.whoopConnection.update({ where: { userId }, diff --git a/src/lib/nightscout/__tests__/sync.test.ts b/src/lib/nightscout/__tests__/sync.test.ts index f9f46043a..770eb67d4 100644 --- a/src/lib/nightscout/__tests__/sync.test.ts +++ b/src/lib/nightscout/__tests__/sync.test.ts @@ -4,6 +4,8 @@ const { fetchSgvEntriesMock, getCredsMock, upsertMock, + findManyMock, + emitArrivalsMock, recordSuccessMock, recordFailureMock, recomputeMock, @@ -12,6 +14,8 @@ const { fetchSgvEntriesMock: vi.fn(), getCredsMock: vi.fn(), upsertMock: vi.fn(), + findManyMock: vi.fn(), + emitArrivalsMock: vi.fn(), recordSuccessMock: vi.fn(), recordFailureMock: vi.fn(), recomputeMock: vi.fn(), @@ -23,7 +27,11 @@ vi.mock("../credentials", () => ({ })); vi.mock("@/lib/db", () => ({ - prisma: { measurement: { upsert: upsertMock } }, + prisma: { measurement: { upsert: upsertMock, findMany: findManyMock } }, +})); + +vi.mock("@/lib/arrivals/measurement-emit", () => ({ + emitInsertedMeasurementArrivals: emitArrivalsMock, })); vi.mock("@/lib/integrations/status", async (importOriginal) => ({ @@ -63,12 +71,17 @@ beforeEach(() => { fetchSgvEntriesMock.mockReset(); getCredsMock.mockReset(); upsertMock.mockReset(); + findManyMock.mockReset(); + emitArrivalsMock.mockReset(); recordSuccessMock.mockReset(); recordFailureMock.mockReset(); recomputeMock.mockReset(); invalidateMock.mockReset(); getCredsMock.mockResolvedValue(CREDS); - upsertMock.mockResolvedValue({}); + upsertMock.mockResolvedValue({ id: "row" }); + // Default: no rows pre-exist → every entry is a fresh insert. + findManyMock.mockResolvedValue([]); + emitArrivalsMock.mockResolvedValue(undefined); invalidateMock.mockResolvedValue(undefined); }); @@ -128,6 +141,51 @@ describe("syncUserNightscout", () => { expect(recordSuccessMock).toHaveBeenCalledWith("u1", "nightscout"); }); + it("N-1 — counts only fresh inserts and emits ONLY the newly-inserted rows", async () => { + fetchSgvEntriesMock.mockResolvedValue([ENTRY_A, ENTRY_B]); + // ENTRY_A already exists (a re-confirm); ENTRY_B is genuinely new. + findManyMock.mockResolvedValue([{ externalId: "ns:abc" }]); + + const n = await syncUserNightscout("u1"); + + // The inflated insert+reconfirm count used to fire the reminder-satisfy + // trigger every tick; the honest count is the single fresh insert. + expect(n).toBe(1); + // Both rows are still written (idempotent), but only the fresh one emits. + expect(upsertMock).toHaveBeenCalledTimes(2); + expect(emitArrivalsMock).toHaveBeenCalledTimes(1); + const [uid, rows, source] = emitArrivalsMock.mock.calls[0]!; + expect(uid).toBe("u1"); + expect(source).toBe("nightscout"); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + type: "BLOOD_GLUCOSE", + measuredAt: new Date(1718000300000), + }); + expect(recordSuccessMock).toHaveBeenCalledWith("u1", "nightscout"); + }); + + it("N-2 — a partial row failure records a transient failure and SKIPS success", async () => { + fetchSgvEntriesMock.mockResolvedValue([ENTRY_A, ENTRY_B]); + findManyMock.mockResolvedValue([]); + upsertMock + .mockResolvedValueOnce({ id: "r1" }) + .mockRejectedValueOnce(new Error("bad row")); + + const n = await syncUserNightscout("u1"); + + // The one good row still landed; the tick is honest about the failed row. + expect(n).toBe(1); + expect(recordSuccessMock).not.toHaveBeenCalled(); + expect(recordFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "nightscout", + kind: "transient", + message: "1 of 2 SGV rows failed to write", + }), + ); + }); + it("records a failure and rethrows when the instance is unreachable", async () => { fetchSgvEntriesMock.mockRejectedValue( new NightscoutApiError("Nightscout responded 401", 401), diff --git a/src/lib/nightscout/sync.ts b/src/lib/nightscout/sync.ts index 0a7ef96f7..44588d512 100644 --- a/src/lib/nightscout/sync.ts +++ b/src/lib/nightscout/sync.ts @@ -23,6 +23,7 @@ import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; import { getEvent } from "@/lib/logging/context"; import { + markSyncFailureRecorded, recordSyncFailure, recordSyncSuccess, toFailureKind, @@ -33,6 +34,10 @@ import { collapseToTypeDayKeys, recomputeBucketsForMeasurement, } from "@/lib/rollups/measurement-rollups"; +import { + emitInsertedMeasurementArrivals, + type InsertedMeasurementArrivalRow, +} from "@/lib/arrivals/measurement-emit"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { fetchSgvEntries, @@ -46,7 +51,12 @@ import { getUserNightscoutCredentials } from "./credentials"; * How many recent SGV entries to pull per sync tick. A CGM emits one reading * every ~5 min (≈288/day); 576 covers ~48 h of catch-up so a missed hourly * tick (or a brief outage) still backfills on the next run without paging the - * full instance. The far-history backfill walks larger windows separately. + * full instance. + * + * This ~48 h window is the ONLY history Nightscout imports: there is no + * far-history backfill queue, so connecting Nightscout brings in about two + * days of readings and every tick thereafter walks the same recent set. A + * deeper walk would need its own windowed-pagination queue; it is not built. */ export const NIGHTSCOUT_SYNC_COUNT = 576; @@ -69,6 +79,32 @@ export function redactNightscoutSecret(message: string): string { return message.replace(/\b(token|api-secret)=[^\s&"']+/gi, "$1=REDACTED"); } +/** + * Record a Nightscout sync failure on the shared `nightscout` ledger with the + * correct classification + HTTP code + the token STRIPPED from the message. + * Extracted so the poll-cohort boundary can record a future escape with the + * same provider-owned classification AND redaction the inline fetch catch + * applies — a boundary-blind recorder would persist the token-bearing URL a + * `SafeFetchError` embeds. + */ +export async function recordNightscoutSyncFailure( + userId: string, + err: unknown, +): Promise { + await recordSyncFailure({ + userId, + integration: "nightscout", + kind: classifyNightscoutFailure(err), + message: redactNightscoutSecret( + err instanceof Error ? err.message : String(err), + ), + errorCode: + err instanceof NightscoutApiError && err.status != null + ? String(err.status) + : undefined, + }); +} + /** Map a Nightscout HTTP status / network error onto the shared ledger kind. */ export function classifyNightscoutFailure(err: unknown): FailureKind { let classification: IntegrationClassification = "transient"; @@ -86,9 +122,12 @@ export function classifyNightscoutFailure(err: unknown): FailureKind { } /** - * Sync one user's recent SGV entries. Returns the count of rows written - * (newly inserted or re-confirmed). A user with no configured instance is a - * clean no-op (returns 0, touches no status row). + * Sync one user's recent SGV entries. Returns the count of rows FRESHLY + * INSERTED this tick (re-confirmed rows the upsert merely touched are not + * counted — the old insert+reconfirm total inflated the cohort's + * `measurements_imported` and fired the reminder-satisfy trigger every tick + * regardless of whether new glucose actually arrived). A user with no + * configured instance is a clean no-op (returns 0, touches no status row). */ export async function syncUserNightscout( userId: string, @@ -106,25 +145,34 @@ export async function syncUserNightscout( allowPrivateHost: creds.allowPrivateHost, }); } catch (err) { + // Recorded here (token-redacted), then marked so the poll-cohort boundary + // does not double-record it — and never re-persists the token-bearing URL + // a boundary-blind recorder would. + await recordNightscoutSyncFailure(userId, err); + throw markSyncFailureRecorded(err); + } + + const { inserted, failedRows } = await upsertNightscoutEntries( + userId, + entries, + ); + + if (failedRows > 0) { + // Partial write failure: keep the tick honest. `transient` because the + // ~48 h re-fetch window self-heals a transient DB hiccup, while a + // persistent malformed-row failure now accrues visible strikes instead of + // being invisibly reset by a success stamp. Do NOT stamp success. await recordSyncFailure({ userId, integration: "nightscout", - kind: classifyNightscoutFailure(err), - message: redactNightscoutSecret( - err instanceof Error ? err.message : String(err), - ), - errorCode: - err instanceof NightscoutApiError && err.status != null - ? String(err.status) - : undefined, + kind: "transient", + message: `${failedRows} of ${entries.length} SGV rows failed to write`, }); - throw err; + return inserted; } - const imported = await upsertNightscoutEntries(userId, entries); - await recordSyncSuccess(userId, "nightscout"); - return imported; + return inserted; } /** @@ -132,38 +180,71 @@ export async function syncUserNightscout( * rollup tier + invalidate status-insight caches once at the end (mirrors the * Withings / WHOOP sync tail). First-write-wins: the `update` branch is empty * so a re-sync of the same reading is a no-op rather than a rewrite. Best-effort - * on the per-row write + the rollup fold — one bad row never aborts the pass. + * on the per-row write + the rollup fold — one bad row never aborts the pass, + * and the count of failed rows is returned so the caller can surface a partial + * write failure instead of stamping success over it. + * + * Returns `inserted` (rows whose key was absent before this pass — the honest + * "new readings" count that also drives the arrival emit) and `failedRows`. */ export async function upsertNightscoutEntries( userId: string, entries: ParsedSgvEntry[], -): Promise { - if (entries.length === 0) return 0; +): Promise<{ inserted: number; failedRows: number }> { + if (entries.length === 0) return { inserted: 0, failedRows: 0 }; const type = "BLOOD_GLUCOSE" as MeasurementType; - let imported = 0; + const mapped = entries.map((entry) => mapSgvEntryToMeasurement(entry)); + + // Probe which keys already exist so a re-confirmed reading (the upsert's + // no-op update branch) is not miscounted as a fresh import and the arrival + // spine emits ONLY the genuinely-new rows. Cohort passes are sequential per + // user, so no concurrent insert races this probe. A probe failure degrades + // to "treat every row as fresh" — the rollup fold + count self-correct on + // the next tick. + const existingKeys = new Set(); + try { + const existing = await prisma.measurement.findMany({ + where: { + userId, + type, + source: "NIGHTSCOUT", + externalId: { in: mapped.map((m) => m.externalId) }, + }, + select: { externalId: true }, + }); + for (const row of existing) { + if (row.externalId) existingKeys.add(row.externalId); + } + } catch (err) { + getEvent()?.addWarning( + `nightscout: insert-probe failed for ${userId}: ${err}`, + ); + } + + let failedRows = 0; + const insertedRows: InsertedMeasurementArrivalRow[] = []; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; - for (const entry of entries) { - const mapped = mapSgvEntryToMeasurement(entry); + for (const m of mapped) { try { - await prisma.measurement.upsert({ + const row = await prisma.measurement.upsert({ where: { userId_type_source_externalId: { userId, type, source: "NIGHTSCOUT", - externalId: mapped.externalId, + externalId: m.externalId, }, }, create: { userId, type, source: "NIGHTSCOUT", - value: mapped.value, - unit: mapped.unit, - measuredAt: mapped.measuredAt, - externalId: mapped.externalId, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.externalId, }, // First-write-wins: a CGM sample is immutable, so a re-sync of the // same reading must not rewrite the stored value or measuredAt. @@ -172,16 +253,32 @@ export async function upsertNightscoutEntries( // truth for its own rows, so a re-synced reading brings the row // back (mirrors Google / Fitbit). update: { deletedAt: null }, + select: { id: true }, }); - touched.push({ type, measuredAt: mapped.measuredAt }); - imported++; + touched.push({ type, measuredAt: m.measuredAt }); + if (!existingKeys.has(m.externalId)) { + insertedRows.push({ type, measuredAt: m.measuredAt, id: row.id }); + } } catch (err) { getEvent()?.addWarning( `nightscout: failed to upsert measurement: ${err}`, ); + failedRows++; } } + // Spine parity with every other measurement writer: emit at most one arrival + // per supported kind for the rows freshly inserted. BLOOD_GLUCOSE is not yet + // an arrival kind (`ARRIVAL_MEASUREMENT_KIND`), so this is behaviourally + // inert for glucose today — it removes Nightscout as the spine's odd one out + // and lights up automatically once a `glucose` arrival kind + its consuming + // surface land in a later release. + void emitInsertedMeasurementArrivals( + userId, + insertedRows, + "nightscout", + ).catch(() => {}); + try { const keys = collapseToTypeDayKeys(touched); for (const k of keys) { @@ -201,5 +298,5 @@ export async function upsertNightscoutEntries( ); } - return imported; + return { inserted: insertedRows.length, failedRows }; } diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index d59602a77..798947ed9 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/src/lib/oura/__tests__/sync.test.ts @@ -114,6 +114,7 @@ vi.mock("../client", async (importOriginal) => { import { syncUserOura, upsertOuraMeasurements } from "../sync"; import { OuraApiError } from "../response-classifier"; +import { isSyncFailureRecorded } from "@/lib/integrations/status"; const CONN = { accessToken: "acc", @@ -284,13 +285,39 @@ describe("syncUserOura", () => { upstreamError: "invalid_grant", }), ); - await expect(syncUserOura("u1")).rejects.toBeInstanceOf(OuraApiError); + const err = await syncUserOura("u1").catch((e) => e); + expect(err).toBeInstanceOf(OuraApiError); + // Recorded at the source AND marked, so the poll-cohort boundary does not + // record it a second time. + expect(isSyncFailureRecorded(err)).toBe(true); + expect(recordFailureMock).toHaveBeenCalledTimes(1); expect(recordFailureMock).toHaveBeenCalledWith( expect.objectContaining({ integration: "oura", kind: "reauth_required" }), ); expect(recordSuccessMock).not.toHaveBeenCalled(); }); + it("leaves a write-path failure UNMARKED so the cohort boundary records it once", async () => { + // The `upsertOuraMeasurements` transaction throw escapes `syncUserOura` + // uncaught: no source-site record, no mark — the poll-cohort boundary is + // what records it (exactly once). + getConnMock.mockResolvedValue(CONN); + fetchReadinessMock.mockResolvedValue([ + { id: "r1", day: "2026-06-10", score: 71 }, + ]); + reconcileMock.mockResolvedValue({ + status: "failed", + reason: "db_error", + error: new Error("write boom"), + }); + + const err = await syncUserOura("u1").catch((e) => e); + expect(err).toBeInstanceOf(Error); + expect(isSyncFailureRecorded(err)).toBe(false); + expect(recordFailureMock).not.toHaveBeenCalled(); + expect(recordSuccessMock).not.toHaveBeenCalled(); + }); + it("keeps a nap and the main sleep distinct via record-scoped externalIds (B2)", async () => { getConnMock.mockResolvedValue(CONN); // Two sleep records on the same day: a main sleep + a nap. diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index 779cf1bdf..9ed4bc39a 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -37,6 +37,7 @@ import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; import { getEvent } from "@/lib/logging/context"; import { + markSyncFailureRecorded, recordSyncFailure, recordSyncSuccess, toFailureKind, @@ -97,6 +98,28 @@ export function classifyOuraFailure(err: unknown): FailureKind { return toFailureKind(classifyOuraError(err)); } +/** + * Record a whole-connection Oura sync failure on the shared `oura` ledger with + * the correct classification + HTTP code. Extracted so the poll-cohort boundary + * can record a future escape with the SAME provider-owned classification the + * inline catch already applies. + */ +export async function recordOuraSyncFailure( + userId: string, + err: unknown, +): Promise { + await recordSyncFailure({ + userId, + integration: "oura", + kind: classifyOuraFailure(err), + message: err instanceof Error ? err.message : String(err), + errorCode: + err instanceof OuraApiError && err.httpStatus != null + ? String(err.httpStatus) + : undefined, + }); +} + function ymd(d: Date): string { return d.toISOString().slice(0, 10); } @@ -355,17 +378,10 @@ export async function syncUserOura( } 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", - kind: classifyOuraFailure(err), - message: err instanceof Error ? err.message : String(err), - errorCode: - err instanceof OuraApiError && err.httpStatus != null - ? String(err.httpStatus) - : undefined, - }); - throw err; + // Recorded here, then marked so the poll-cohort boundary does not + // double-record it when it surfaces there. + await recordOuraSyncFailure(userId, err); + throw markSyncFailureRecorded(err); } // Clear whatever an earlier scoring left under the re-fetched sleep records diff --git a/src/lib/polar/__tests__/sync.test.ts b/src/lib/polar/__tests__/sync.test.ts index 9b717ee72..71aace68c 100644 --- a/src/lib/polar/__tests__/sync.test.ts +++ b/src/lib/polar/__tests__/sync.test.ts @@ -92,6 +92,7 @@ vi.mock("../client", async (importOriginal) => { }); import { syncUserPolar, upsertPolarMeasurements } from "../sync"; +import { isSyncFailureRecorded } from "@/lib/integrations/status"; import { PolarApiError } from "../response-classifier"; const CONN = { accessToken: "tok", polarUserId: "42" }; @@ -340,7 +341,12 @@ describe("syncUserPolar", () => { reason: "http_401", }), ); - await expect(syncUserPolar("u1")).rejects.toBeInstanceOf(PolarApiError); + 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); + expect(recordFailureMock).toHaveBeenCalledTimes(1); expect(recordFailureMock).toHaveBeenCalledWith( expect.objectContaining({ userId: "u1", @@ -351,6 +357,27 @@ describe("syncUserPolar", () => { ); expect(recordSuccessMock).not.toHaveBeenCalled(); }); + + 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 + // what records it (exactly once). + getConnMock.mockResolvedValue(CONN); + fetchRechargesMock.mockResolvedValue([ + { date: "2026-06-10", nightly_recharge_status: 6 }, + ]); + reconcileMock.mockReset().mockResolvedValue({ + status: "failed", + reason: "db_error", + error: new Error("write boom"), + }); + + const err = await syncUserPolar("u1").catch((e) => e); + expect(err).toBeInstanceOf(Error); + expect(isSyncFailureRecorded(err)).toBe(false); + expect(recordFailureMock).not.toHaveBeenCalled(); + expect(recordSuccessMock).not.toHaveBeenCalled(); + }); }); describe("upsertPolarMeasurements — exact insertion results", () => { diff --git a/src/lib/polar/sync-workouts.ts b/src/lib/polar/sync-workouts.ts index 5893e31d8..0f7e942c0 100644 --- a/src/lib/polar/sync-workouts.ts +++ b/src/lib/polar/sync-workouts.ts @@ -31,7 +31,11 @@ import { prisma } from "@/lib/db"; import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit"; import { annotate, getEvent } from "@/lib/logging/context"; -import { recordSyncFailure, toFailureKind } from "@/lib/integrations/status"; +import { + markSyncFailureRecorded, + recordSyncFailure, + toFailureKind, +} from "@/lib/integrations/status"; import { fetchExercises, mapExercise, type PolarWorkoutRow } from "./client"; import { getPolarConnection } from "./credentials"; import { PolarApiError, classifyPolarError } from "./response-classifier"; @@ -82,7 +86,7 @@ export async function syncUserPolarWorkouts(userId: string): Promise { } } catch (err) { await recordPolarWorkoutFailure(userId, err); - throw err; + throw markSyncFailureRecorded(err); } let imported: number; @@ -90,7 +94,7 @@ export async function syncUserPolarWorkouts(userId: string): Promise { imported = await upsertPolarWorkouts(userId, rows); } catch (err) { await recordPolarWorkoutFailure(userId, err); - throw err; + throw markSyncFailureRecorded(err); } return imported; diff --git a/src/lib/polar/sync.ts b/src/lib/polar/sync.ts index f43c184d9..7878c4475 100644 --- a/src/lib/polar/sync.ts +++ b/src/lib/polar/sync.ts @@ -25,6 +25,7 @@ import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; import { getEvent } from "@/lib/logging/context"; import { + markSyncFailureRecorded, recordSyncFailure, recordSyncSuccess, toFailureKind, @@ -69,6 +70,28 @@ export function classifyPolarFailure(err: unknown): FailureKind { return toFailureKind(classifyPolarError(err)); } +/** + * Record a Polar vitals-leg sync failure on the shared `polar` ledger with the + * correct classification + HTTP code. Extracted so the poll-cohort boundary can + * record a future escape with the SAME provider-owned classification the inline + * catch already applies. + */ +export async function recordPolarSyncFailure( + userId: string, + err: unknown, +): Promise { + await recordSyncFailure({ + userId, + integration: "polar", + kind: classifyPolarFailure(err), + message: err instanceof Error ? err.message : String(err), + errorCode: + err instanceof PolarApiError && err.httpStatus != null + ? String(err.httpStatus) + : undefined, + }); +} + /** One mapped reading with its `externalId` resolved. */ export interface PolarMeasurementUpsert { type: string; @@ -151,17 +174,10 @@ export async function syncUserPolar(userId: string): Promise { readings.push(...toUpsert(mapSpo2(t), "spo2")); } } catch (err) { - await recordSyncFailure({ - userId, - integration: "polar", - kind: classifyPolarFailure(err), - message: err instanceof Error ? err.message : String(err), - errorCode: - err instanceof PolarApiError && err.httpStatus != null - ? String(err.httpStatus) - : undefined, - }); - throw 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 diff --git a/src/lib/strava/__tests__/sync-cursor-durability.test.ts b/src/lib/strava/__tests__/sync-cursor-durability.test.ts index 18019f19a..dc71edf55 100644 --- a/src/lib/strava/__tests__/sync-cursor-durability.test.ts +++ b/src/lib/strava/__tests__/sync-cursor-durability.test.ts @@ -39,6 +39,7 @@ vi.mock("@/lib/integrations/status", () => ({ recordSyncFailure: mocks.recordSyncFailure, recordSyncSuccess: mocks.recordSyncSuccess, toFailureKind: vi.fn(() => "transient"), + markSyncFailureRecorded: (err: T) => err, })); vi.mock("../credentials", () => ({ getStravaConnection: mocks.getStravaConnection, diff --git a/src/lib/strava/sync.ts b/src/lib/strava/sync.ts index e416407c4..346850464 100644 --- a/src/lib/strava/sync.ts +++ b/src/lib/strava/sync.ts @@ -27,6 +27,7 @@ import { prisma } from "@/lib/db"; import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit"; import { annotate, getEvent } from "@/lib/logging/context"; import { + markSyncFailureRecorded, recordSyncFailure, recordSyncSuccess, toFailureKind, @@ -68,7 +69,7 @@ export function classifyStravaFailure(err: unknown): FailureKind { return toFailureKind(classifyStravaError(err)); } -async function recordStravaFailure( +export async function recordStravaFailure( userId: string, err: unknown, ): Promise { @@ -215,7 +216,7 @@ export async function syncUserStrava( } } catch (err) { await recordStravaFailure(userId, err); - throw err; + throw markSyncFailureRecorded(err); } let imported: number; @@ -239,7 +240,7 @@ export async function syncUserStrava( } } catch (err) { await recordStravaFailure(userId, err); - throw err; + throw markSyncFailureRecorded(err); } await recordSyncSuccess(userId, "strava"); diff --git a/src/lib/whoop/__tests__/sync-cycle-overlap.test.ts b/src/lib/whoop/__tests__/sync-cycle-overlap.test.ts new file mode 100644 index 000000000..27b865782 --- /dev/null +++ b/src/lib/whoop/__tests__/sync-cycle-overlap.test.ts @@ -0,0 +1,83 @@ +/** + * Q8 — WHOOP cycle ingest used the narrow 1 h default overlap while workout / + * recovery / sleep use the 7-day re-score overlap. A cycle re-scored through the + * day (or fetched just after the cursor stepped past it) then fell before the + * incremental `start` and its final value was never re-fetched. This pins cycle + * onto the SAME overlap the other collections use (the upsert is idempotent). + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const fetchCycles = vi.fn(); + +vi.mock("../client", () => ({ + fetchCycles: (...a: unknown[]) => fetchCycles(...a), + mapCycle: () => [], +})); + +const getValidToken = vi.fn(); +const markResourceSynced = vi.fn(); +const upsertWhoopMeasurements = vi.fn(); +vi.mock("../sync-core", async () => { + const actual = + await vi.importActual("../sync-core"); + return { + ...actual, + getValidToken: (...a: unknown[]) => getValidToken(...a), + markResourceSynced: (...a: unknown[]) => markResourceSynced(...a), + upsertWhoopMeasurements: (...a: unknown[]) => upsertWhoopMeasurements(...a), + }; +}); + +const whoopConnectionFindUnique = vi.fn(); +vi.mock("@/lib/db", () => ({ + prisma: { + whoopConnection: { + findUnique: (...a: unknown[]) => whoopConnectionFindUnique(...a), + }, + }, +})); + +import { syncUserCycle } from "../sync-cycle"; +import { + WHOOP_FULL_SYNC_ANCHOR, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, +} from "../sync-core"; + +const LAST_SYNCED = new Date("2026-06-10T12:00:00.000Z"); + +beforeEach(() => { + vi.clearAllMocks(); + getValidToken.mockResolvedValue({ + accessToken: "tok", + connection: { id: "c1", whoopUserId: "w1" }, + }); + whoopConnectionFindUnique.mockResolvedValue({ + lastSyncedAt: LAST_SYNCED, + resourceCursors: null, + }); + markResourceSynced.mockResolvedValue(undefined); + upsertWhoopMeasurements.mockResolvedValue(0); + fetchCycles.mockResolvedValue([]); +}); + +describe("syncUserCycle — 7-day re-score overlap window", () => { + it("fetches from lastSyncedAt minus the 7-day overlap, not 1 h", async () => { + await syncUserCycle("user-1"); + + expect(fetchCycles).toHaveBeenCalledTimes(1); + const arg = fetchCycles.mock.calls[0]![1] as { start?: Date }; + expect(arg.start?.getTime()).toBe( + LAST_SYNCED.getTime() - WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + ); + // A cycle still settling 6 h after the cursor is inside the fetch range; + // under the old 1 h overlap it would be missed. + const sixHoursBefore = LAST_SYNCED.getTime() - 6 * 60 * 60 * 1000; + expect(arg.start!.getTime()).toBeLessThanOrEqual(sixHoursBefore); + }); + + it("fullSync anchors at the deep-history anchor, not the cursor", async () => { + await syncUserCycle("user-1", { fullSync: true }); + const arg = fetchCycles.mock.calls[0]![1] as { start?: Date }; + expect(arg.start).toEqual(WHOOP_FULL_SYNC_ANCHOR); + }); +}); diff --git a/src/lib/whoop/__tests__/sync-dead-token.test.ts b/src/lib/whoop/__tests__/sync-dead-token.test.ts new file mode 100644 index 000000000..3065f74a2 --- /dev/null +++ b/src/lib/whoop/__tests__/sync-dead-token.test.ts @@ -0,0 +1,277 @@ +/** + * WH-1 — pins the dead-token verdict for WHOOP, the parity mirror of + * `fitbit/__tests__/sync-dead-token.test.ts`. + * + * `getValidToken`'s three null-return paths (credentials missing, refresh + * failure inside the advisory-lock transaction, and the in-lock vanished + * connection) must register on the cycle's hard-fail ledger, and a cycle whose + * token is dead must NOT stamp `recordSyncSuccess`. Otherwise every resource + * returns 0 without failures, the cycle reads as clean, `recordSyncSuccess` + * un-parks `error_reauth`, and the hourly cohort hammers the dead refresh token + * forever while `lastSyncedAt` advances past real data. + * + * Includes the WHOOP-specific pin Fitbit has no twin for: the per-resource + * driver `syncWhoopResourceWithStatus` (the dominant hourly-cron path) must + * also refuse the success stamp when a hard failure is noted inside `run`. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { + recordSyncFailureMock, + recordSyncSuccessMock, + isReauthRequiredMock, + prismaMock, + getUserWhoopCredentialsMock, + refreshAccessTokenMock, + resourceFake, +} = vi.hoisted(() => ({ + recordSyncFailureMock: vi.fn(async () => {}), + recordSyncSuccessMock: vi.fn(async () => {}), + isReauthRequiredMock: vi.fn(async () => false), + prismaMock: { + whoopConnection: { + findUnique: vi.fn(), + update: vi.fn(async () => ({})), + }, + $transaction: vi.fn(), + }, + getUserWhoopCredentialsMock: vi.fn(async (): Promise => null), + refreshAccessTokenMock: vi.fn(), + // The dead-token cycle test drives the real `syncUserWhoop`; each resource + // module resolves the token first — exactly what the real leaves do — so the + // ledger registration happens inside the cycle's ALS scope. + resourceFake: async (userId: string): Promise => { + const { getValidToken } = await import("../sync-core"); + return (await getValidToken(userId)) ? 1 : 0; + }, +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/crypto", () => ({ + encrypt: (s: string) => `enc(${s})`, + decrypt: (s: string) => s.replace(/^enc\(|\)$/g, ""), +})); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: isReauthRequiredMock, + recordSyncFailure: recordSyncFailureMock, + recordSyncSuccess: recordSyncSuccessMock, + isSyncFailureRecorded: () => false, + markSyncFailureRecorded: (err: T) => err, +})); +vi.mock("@/lib/integrations/oauth-refresh", () => ({ + acquireProviderTokenRefreshLock: vi.fn(async () => {}), + PROVIDER_REFRESH_TRANSACTION_OPTIONS: { maxWait: 10_000, timeout: 60_000 }, +})); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: vi.fn(), + MeasurementReconciliationError: class extends Error {}, +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: vi.fn(() => []), + recomputeBucketsForMeasurement: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); +vi.mock("@/lib/arrivals/measurement-emit", () => ({ + emitInsertedMeasurementArrivals: vi.fn(async () => {}), +})); +vi.mock("@/lib/logging/context", () => ({ + getEvent: () => null, + annotate: () => {}, +})); +vi.mock("../credentials", () => ({ + getUserWhoopCredentials: getUserWhoopCredentialsMock, +})); +vi.mock("../client", async (orig) => { + const actual = await orig(); + return { ...actual, refreshAccessToken: refreshAccessTokenMock }; +}); + +// Each resource resolves the token first (the ledger-registration seam). +vi.mock("../sync-recovery", () => ({ syncUserRecovery: resourceFake })); +vi.mock("../sync-sleep", () => ({ syncUserSleep: resourceFake })); +vi.mock("../sync-cycle", () => ({ syncUserCycle: resourceFake })); +vi.mock("../sync-workout", () => ({ syncUserWorkout: resourceFake })); +vi.mock("../sync-body", () => ({ syncUserBody: resourceFake })); + +import { + WHOOP_TOKEN_HARD_FAIL, + getValidToken, + hardFailStorage, + noteHardFailure, + syncWhoopResourceWithStatus, +} from "../sync-core"; +import { syncUserWhoop } from "../sync"; +import { WhoopApiError } from "../response-classifier"; + +/** Run `fn` inside a fresh hard-fail ledger scope and return its failures. */ +async function withLedger( + fn: () => Promise, +): Promise<{ result: T; failures: string[] }> { + const tracker = { failures: [] as string[] }; + const result = await hardFailStorage.run(tracker, fn); + return { result, failures: tracker.failures }; +} + +/** A connection whose access token is inside the 5-min refresh buffer. */ +const EXPIRED_CONNECTION = { + id: "conn-1", + whoopUserId: "wu-1", + accessToken: "enc(access)", + refreshToken: "enc(refresh)", + tokenExpiresAt: new Date(Date.now() - 60_000), + lastSyncedAt: new Date("2026-07-01T00:00:00.000Z"), +}; + +const CREDS = { clientId: "id", clientSecret: "secret" }; + +beforeEach(() => { + vi.clearAllMocks(); + isReauthRequiredMock.mockResolvedValue(false); + prismaMock.whoopConnection.findUnique.mockReset(); + prismaMock.whoopConnection.update.mockReset().mockResolvedValue({}); + getUserWhoopCredentialsMock.mockReset().mockResolvedValue(null); + refreshAccessTokenMock.mockReset(); + // The advisory-lock transaction runs its callback against the same mock. + prismaMock.$transaction + .mockReset() + .mockImplementation(async (run: (tx: unknown) => unknown) => + run(prismaMock), + ); +}); + +describe("getValidToken — dead-token cycle verdict", () => { + it("registers the credentials-missing path on the hard-fail ledger", async () => { + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(null); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([WHOOP_TOKEN_HARD_FAIL]); + expect(recordSyncFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "whoop", + kind: "reauth_required", + errorCode: "credentials_missing", + }), + ); + }); + + it("registers a refresh failure on the hard-fail ledger", async () => { + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockRejectedValue( + new WhoopApiError({ + verb: "refreshToken", + classification: "reauth_required", + httpStatus: 401, + reason: "invalid_grant", + }), + ); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([WHOOP_TOKEN_HARD_FAIL]); + expect(recordSyncFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "whoop", + kind: "reauth_required", + }), + ); + }); + + it("registers an in-lock vanished connection on the ledger", async () => { + // The connection was deleted between the first read and the in-lock re-read + // (disconnect / webhook race). + prismaMock.whoopConnection.findUnique + .mockResolvedValueOnce(EXPIRED_CONNECTION as never) // initial read + .mockResolvedValueOnce(null as never); // in-lock re-read: row gone + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([WHOOP_TOKEN_HARD_FAIL]); + expect(refreshAccessTokenMock).not.toHaveBeenCalled(); + }); + + it("a successful refresh leaves the ledger empty", async () => { + // Guard against over-failing: a healthy refresh must not touch the ledger. + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockResolvedValue({ + access_token: "new-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result?.accessToken).toBe("new-access"); + expect(failures).toEqual([]); + }); +}); + +describe("syncUserWhoop — dead-token cycle refuses the success stamp", () => { + it("a dead-token cycle returns { imported: 0, failed: true } and never stamps success", async () => { + // Every resource resolves the token; the refresh fails each time — before + // the ledger fix the cycle read as CLEAN (all resources returned 0 without + // throwing), stamped success, and un-parked error_reauth. + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockRejectedValue( + new WhoopApiError({ + verb: "refreshToken", + classification: "reauth_required", + httpStatus: 401, + reason: "invalid_grant", + }), + ); + + const res = await syncUserWhoop("user-1"); + + expect(res).toEqual({ imported: 0, failed: true }); + expect(recordSyncSuccessMock).not.toHaveBeenCalled(); + }); +}); + +describe("syncWhoopResourceWithStatus — dominant per-resource cron path", () => { + it("does NOT stamp success when a hard failure is noted inside run", async () => { + // The hourly-cron arm Fitbit has no twin for: a dead token noted by + // `getValidToken` inside `run` must block the per-resource success stamp, + // or the release is inert on the path that runs most often. + const imported = await syncWhoopResourceWithStatus("user-1", async () => { + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); + return 0; + }); + + expect(imported).toBe(0); + expect(recordSyncSuccessMock).not.toHaveBeenCalled(); + }); + + it("stamps success on a clean per-resource run (no hard failure)", async () => { + const imported = await syncWhoopResourceWithStatus("user-1", async () => 3); + + expect(imported).toBe(3); + expect(recordSyncSuccessMock).toHaveBeenCalledWith("user-1", "whoop"); + }); +}); diff --git a/src/lib/whoop/__tests__/sync-user-whoop.test.ts b/src/lib/whoop/__tests__/sync-user-whoop.test.ts index a1ebab5b9..2c5fcfbdc 100644 --- a/src/lib/whoop/__tests__/sync-user-whoop.test.ts +++ b/src/lib/whoop/__tests__/sync-user-whoop.test.ts @@ -46,6 +46,8 @@ vi.mock("@/lib/integrations/status", () => ({ recordSyncFailure: (...a: unknown[]) => recordSyncFailure(...a), recordSyncSuccess: (...a: unknown[]) => recordSyncSuccess(...a), isReauthRequired: (...a: unknown[]) => isReauthRequired(...a), + isSyncFailureRecorded: () => false, + markSyncFailureRecorded: (err: T) => err, })); vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: () => [], @@ -109,7 +111,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { fn.mockImplementation(softSkip403); } - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(0); expect(recordSyncSuccess).not.toHaveBeenCalled(); @@ -125,7 +127,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { syncUserWorkout.mockImplementation(softSkip403); syncUserBody.mockImplementation(softSkip403); - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(3); expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop"); @@ -138,7 +140,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { syncUserWorkout.mockResolvedValue(0); syncUserBody.mockResolvedValue(0); - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(3); expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop"); @@ -155,7 +157,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { fn.mockResolvedValue(0); } - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(0); // No collection was 403'd → this is a genuine "nothing changed" tick. @@ -165,7 +167,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { it("short-circuits without stamping success when parked at error_reauth", async () => { isReauthRequired.mockResolvedValue(true); - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(0); expect(recordSyncSuccess).not.toHaveBeenCalled(); diff --git a/src/lib/whoop/sync-core.ts b/src/lib/whoop/sync-core.ts index 9c89cdf36..dc54db379 100644 --- a/src/lib/whoop/sync-core.ts +++ b/src/lib/whoop/sync-core.ts @@ -30,6 +30,7 @@ import type { MeasurementType } from "@/generated/prisma/client"; import { encrypt, decrypt } from "@/lib/crypto"; import { getEvent } from "@/lib/logging/context"; import { + markSyncFailureRecorded, recordSyncFailure, recordSyncSuccess, type FailureKind, @@ -97,6 +98,42 @@ export async function runWithWhoopSoftSkipTracking( return { result, softSkipCount: tracker.count }; } +/** + * Per-cycle ledger of HARD failures that must fail the cycle's verdict WITHOUT + * aborting sibling resources — a dead token (`getValidToken` returning null on + * a missing-credentials, failed-refresh, or in-lock vanished-connection path). + * Every resource leaf short-circuits `if (!tokenInfo) return 0` on a null + * token, so without this ledger a dead-token cycle reads as CLEAN: every + * resource "succeeds" with 0 rows, no failure is recorded, and + * `recordSyncSuccess` un-parks `error_reauth`, resets the failure buckets + + * `alertedAt`, and the hourly cohort hammers the dead refresh token forever. + * The orchestrator + the per-resource driver read this ledger to refuse the + * success stamp. Ported from Fitbit R2 / Google Health's `hardFailStorage`. + * Scoped by AsyncLocalStorage so concurrent per-user jobs never cross-pollinate. + */ +export interface HardFailTracker { + failures: string[]; +} +export const hardFailStorage = new AsyncLocalStorage(); + +/** + * Ledger label for a dead-token failure (`getValidToken` returning null on a + * missing-credentials, failed-refresh, or in-lock vanished-connection path). + * Ported from Fitbit's `FITBIT_TOKEN_HARD_FAIL`. + */ +export const WHOOP_TOKEN_HARD_FAIL = "token"; + +/** + * Record a hard failure on the ambient per-cycle ledger, if one is in scope. + * No-op outside a ledger scope (e.g. a direct REPL call). Used by + * `getValidToken`'s dead-token paths — anything that must fail the cycle's + * verdict without aborting sibling work. Ported from Fitbit's `noteHardFailure`. + */ +export function noteHardFailure(label: string): void { + const tracker = hardFailStorage.getStore(); + if (tracker) tracker.failures.push(label); +} + /** * Single-source the per-resource collection-fetch error handling. A 403 on one * data class soft-skips it (warn + return 0) so sibling resources still sync; @@ -122,7 +159,9 @@ export async function handleCollectionFetchError( return 0; } await recordWhoopSyncFailure(userId, err); - throw err; + // Marked so the orchestrator / cron catch (which records unmarked write-path + // throws for WH-4) does not record this fetch failure a SECOND time. + throw markSyncFailureRecorded(err); } /** Refresh the access token this many ms before `tokenExpiresAt`. */ @@ -137,8 +176,14 @@ const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; * narrow overlap; the night's per-stage rows then never reach the DB and * every sleep surface keeps showing the parallel coarse source for that * night. Seven days re-fetches a handful of records per tick (the upserts - * are idempotent) and closes the gap for every realistic lag. Workout/cycle - * settle fast — a smaller overlap suffices and keeps the page count down. + * are idempotent) and closes the gap for every realistic lag. Workout AND + * cycle now ride the same 7-day overlap: a workout can reach the WHOOP cloud + * days after a late phone sync, and a cycle is re-scored through the day — + * both collections filter on the record's own time range, so a narrow overlap + * steps the cursor past a still-settling record and misses it. One record per + * day keeps the re-fetch to a single page. `WHOOP_DEFAULT_OVERLAP_MS` stays + * the narrow fallback `incrementalStart` uses when no resource passes an + * explicit overlap. */ export const WHOOP_RECOVERY_SLEEP_OVERLAP_MS = 7 * 24 * 60 * 60 * 1000; export const WHOOP_DEFAULT_OVERLAP_MS = 60 * 60 * 1000; // 1 h @@ -242,6 +287,11 @@ export async function getValidToken( message: "WHOOP credentials missing — token refresh skipped", errorCode: "credentials_missing", }); + // A dead token means every resource "succeeds" with 0 rows — without a + // ledger entry the cycle reads as clean and `recordSyncSuccess` un-parks + // `error_reauth` while the hourly cohort hammers the dead refresh token. + // Fail the cycle's verdict so the park holds. + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); return null; } @@ -252,7 +302,14 @@ export async function getValidToken( const current = await tx.whoopConnection.findUnique({ where: { userId }, }); - if (!current?.whoopUserId) return null; + if (!current?.whoopUserId) { + // The connection vanished on the re-read inside the advisory lock + // (disconnect/webhook race). Register on the ledger so the cycle does + // not stamp success over a connection that no longer exists — the same + // invariant as the refresh branch below. + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); + return null; + } if ( current.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS >= @@ -295,6 +352,9 @@ export async function getValidToken( `WHOOP token refresh failed for user ${userId}: ${err}`, ); await recordWhoopSyncFailure(userId, err); + // The WH-1 kill shot: an all-zeros refresh-failure run must not stamp + // success, un-park `error_reauth`, and re-arm the alert. Fail the verdict. + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); return null; } } @@ -535,10 +595,18 @@ export async function syncWhoopResourceWithStatus( userId: string, run: () => Promise, ): Promise { + // Run under BOTH trackers: soft-skip (a 403 scope gate) and hard-fail (a dead + // token noted by `getValidToken`). This is the dominant path — the four + // hourly crons walk every connection through here — so without the hard-fail + // gate a dead-token tick would stamp success, un-park `error_reauth`, and the + // whole release would be inert on the path that runs most often. + const hardFailTracker: HardFailTracker = { failures: [] }; const { result: imported, softSkipCount } = - await runWithWhoopSoftSkipTracking(run); + await runWithWhoopSoftSkipTracking(() => + hardFailStorage.run(hardFailTracker, run), + ); const softSkippedOnly = softSkipCount > 0 && imported === 0; - if (!softSkippedOnly) { + if (!softSkippedOnly && hardFailTracker.failures.length === 0) { await recordSyncSuccess(userId, "whoop"); } return imported; diff --git a/src/lib/whoop/sync-cycle.ts b/src/lib/whoop/sync-cycle.ts index c4068ca6f..39c0488ad 100644 --- a/src/lib/whoop/sync-cycle.ts +++ b/src/lib/whoop/sync-cycle.ts @@ -13,6 +13,7 @@ import { markResourceSynced, resolveResourceCursor, upsertWhoopMeasurements, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, type WhoopMeasurementUpsert, } from "./sync-core"; import { prisma } from "@/lib/db"; @@ -30,8 +31,16 @@ export async function syncUserCycle( }); if (!connection) return 0; + // WHOOP re-scores a cycle THROUGH the day (day strain + energy accrue and + // settle after the fact) and the collection filters on the cycle's own time + // range, not on when it updated — so a narrow 1 h overlap could step the + // cursor past a cycle still being revised and miss the final value. Ride the + // same 7-day re-score overlap workout/recovery/sleep already use; one cycle + // per day means ~7 records re-fetched per tick (one page), and the + // `cycle::` upsert keeps the overwrite idempotent. const start = incrementalStart(resolveResourceCursor(connection, "cycle"), { fullSync: opts.fullSync, + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, }); let records: Awaited>; diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index a7e74bc0c..06855227c 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -6,13 +6,22 @@ * the static leaf graph and the full per-user sequencing policy. */ import { getEvent } from "@/lib/logging/context"; -import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status"; +import { + isReauthRequired, + isSyncFailureRecorded, + recordSyncSuccess, +} from "@/lib/integrations/status"; import { syncUserRecovery } from "./sync-recovery"; import { syncUserSleep } from "./sync-sleep"; import { syncUserCycle } from "./sync-cycle"; import { syncUserWorkout } from "./sync-workout"; import { syncUserBody } from "./sync-body"; -import { runWithWhoopSoftSkipTracking } from "./sync-core"; +import { + hardFailStorage, + recordWhoopSyncFailure, + runWithWhoopSoftSkipTracking, + type HardFailTracker, +} from "./sync-core"; /** * Full per-user sync across every WHOOP resource. Webhook-driven syncs enqueue @@ -20,17 +29,18 @@ import { runWithWhoopSoftSkipTracking } from "./sync-core"; * manual `/api/whoop/sync` trigger. * * Parks immediately when the connection is at `error_reauth` (the user must - * reconnect first) — returns 0, matching the Withings no-op contract. + * reconnect first) — returns `{ imported: 0, failed: true }`, matching the + * landed Fitbit verdict contract so the backfill caller can gate its marker. */ export async function syncUserWhoop( userId: string, opts: { fullSync?: boolean } = {}, -): Promise { +): Promise<{ imported: number; failed: boolean }> { if (await isReauthRequired(userId, "whoop")) { getEvent()?.addWarning( `whoop sync skipped for ${userId}: parked at error_reauth`, ); - return 0; + return { imported: 0, failed: true }; } const resources = [ @@ -42,23 +52,41 @@ export async function syncUserWhoop( ]; let anyFailed = false; + // Nest the hard-fail ledger inside the soft-skip tracker (manual nest, not a + // combined wrapper) so a dead token noted by `getValidToken` — or a marked + // fetch hard-fail — fails the verdict just like a thrown resource error. + const hardFailTracker: HardFailTracker = { failures: [] }; const { result: total, softSkipCount } = await runWithWhoopSoftSkipTracking( - async () => { - let imported = 0; - for (const fn of resources) { - try { - imported += await fn(userId, opts); - } catch (err) { - anyFailed = true; - getEvent()?.addWarning( - `whoop ${fn.name} failed for ${userId}: ${err}`, - ); + () => + hardFailStorage.run(hardFailTracker, async () => { + let imported = 0; + for (const fn of resources) { + try { + imported += await fn(userId, opts); + } catch (err) { + anyFailed = true; + // WH-4: a write-path throw (`MeasurementReconciliationError` out of + // `upsertWhoopMeasurements`) reaches no ledger otherwise — a + // stale-green pill during a persistent write failure. Fetch + // hard-fails already recorded + marked in `handleCollectionFetchError`, + // so record only the UNMARKED escapes here (no double-record). + if (!isSyncFailureRecorded(err)) { + await recordWhoopSyncFailure(userId, err); + } + getEvent()?.addWarning( + `whoop ${fn.name} failed for ${userId}: ${err}`, + ); + } } - } - return imported; - }, + return imported; + }), ); + // A dead token / vanished connection notes the hard-fail ledger without + // throwing (each leaf short-circuits `if (!tokenInfo) return 0`); fold that + // into the verdict so an all-zeros dead-token cycle does not stamp success. + anyFailed = anyFailed || hardFailTracker.failures.length > 0; + // A genuine grant-revoke 403s EVERY collection: each resource soft-skips // (returns 0, records no failure), so `anyFailed` stays false and `total` is // 0 — yet the connection is dead until the token-refresh path next catches @@ -71,5 +99,5 @@ export async function syncUserWhoop( if (!anyFailed && !allSoftSkipped) { await recordSyncSuccess(userId, "whoop"); } - return total; + return { imported: total, failed: anyFailed || allSoftSkipped }; } diff --git a/src/lib/withings/__tests__/sync-activity.test.ts b/src/lib/withings/__tests__/sync-activity.test.ts index c34b0d512..0f9c2a80d 100644 --- a/src/lib/withings/__tests__/sync-activity.test.ts +++ b/src/lib/withings/__tests__/sync-activity.test.ts @@ -373,6 +373,31 @@ describe("syncUserActivity — field mapping + idempotency", () => { expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); }); + it("W-2 — records a transient failure and refuses the success stamp when the batch write throws", async () => { + // A swallowed batch-write used to fall through to recordSyncSuccess: green + // pill over lost data, and a PERSISTENT write bug green forever. The write + // failure now records a transient failure and the success stamp is skipped. + installFetchMock([ + { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 }, + ]); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockRejectedValue( + new Error("db write blew up"), + ); + + // No rethrow — sibling chunks + the rollup fold still run (matches Google). + await expect(syncUserActivity("user-1")).resolves.toBeGreaterThanOrEqual(0); + + expect(recordSyncSuccess).not.toHaveBeenCalled(); + expect(recordSyncFailure).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "withings", + kind: "transient", + message: expect.stringContaining("activity batch write failed"), + }), + ); + }); + it("tags every row with the canonical externalId so future replays dedup", async () => { installFetchMock([{ date: "2026-05-12", steps: 8420 }]); vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); diff --git a/src/lib/withings/__tests__/sync-sleep.test.ts b/src/lib/withings/__tests__/sync-sleep.test.ts index 565455c02..98fcff22d 100644 --- a/src/lib/withings/__tests__/sync-sleep.test.ts +++ b/src/lib/withings/__tests__/sync-sleep.test.ts @@ -473,6 +473,32 @@ describe("syncUserSleep — segment writes + idempotency", () => { expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); }); + it("W-3 — records a transient failure and rethrows when a sleep write throws", async () => { + // A reconcile throw out of the stage transaction used to propagate past the + // success stamp and get swallowed by both callers: neither success NOR + // failure recorded, `lastAttemptAt` quietly stale. Now the source records a + // transient failure and rethrows (callers keep their warn-and-continue). + installFetchMock([ + { startdate: 1715000000, enddate: 1715003600, state: 2, id: 1 }, + ]); + reconcileMock.mockReset().mockResolvedValueOnce({ + status: "failed", + reason: "db_error", + error: new Error("reconcile boom"), + }); + + await expect(syncUserSleep("user-1")).rejects.toThrow(); + + expect(recordSyncFailure).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "withings", + kind: "transient", + message: expect.stringContaining("sleep write failed"), + }), + ); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + }); + it("upserts nightly vitals with stable per-vital externalIds (v1.18.10 P0)", async () => { const nightEnd = Math.floor(Date.UTC(2026, 4, 13, 6) / 1000); installSegmentThenSummaryFetch( diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index ca21ad6ca..05431f7ef 100644 --- a/src/lib/withings/sync-activity.ts +++ b/src/lib/withings/sync-activity.ts @@ -311,6 +311,11 @@ export async function syncUserActivity( } let imported = 0; + // W-2 — a swallowed batch-write must not read as a clean sync. The catch + // below sets this so the tail stamps success ONLY when the write landed; a + // persistent write bug then accrues strikes instead of being reset by its + // own success stamp every tick. + let writeFailed = false; // v1.4.39.1 — track every (type, measuredAt) we touched so the // persistent rollup tier can be re-folded at the end of the sync. // See sync.ts header for the full rationale; the chart's @@ -446,9 +451,24 @@ export async function syncUserActivity( ); } } catch (err) { + // Data is lost for this tick and the pill would otherwise read green. Keep + // the no-rethrow (sibling day-chunks + the rollup fold still run) but + // record a transient failure and refuse the success stamp below. + // `transient` because the next hourly tick re-walks the fixed 30-day + // window, so a transient DB error self-heals while a persistent one now + // surfaces on the status card instead of resetting itself. + writeFailed = true; getEvent()?.addWarning( `Failed to batch-upsert ${planned.length} activity rows for ${userId}: ${err}`, ); + await recordSyncFailure({ + userId, + integration: "withings", + kind: "transient", + message: `activity batch write failed: ${ + err instanceof Error ? err.message : String(err) + }`, + }); } } @@ -477,6 +497,10 @@ export async function syncUserActivity( ); } - await recordSyncSuccess(userId, "withings"); + // W-2 — only a clean write stamps success. A swallowed batch-write left the + // ledger honest via the transient failure recorded above. + if (!writeFailed) { + await recordSyncSuccess(userId, "withings"); + } return imported; } diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index f878d39a1..37ba27262 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/src/lib/withings/sync-sleep.ts @@ -412,51 +412,74 @@ export async function syncUserSleep( // and are excluded (their `no-id` prefix would span every id-less night in // history, not just this fetch). const freshBySession = new Map(); - await prisma.$transaction( - async (tx) => { - for (const segment of segments) { - const stage = mapWithingsSleepState(segment.state); - if (!stage) continue; - const measuredAt = new Date(segment.enddate * 1000); - const durationSec = Math.max(0, segment.enddate - segment.startdate); - const minutes = Math.round(durationSec / 60); - const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`; - if (typeof segment.id === "number") { - const fresh = freshBySession.get(segment.id) ?? []; - fresh.push(externalId); - freshBySession.set(segment.id, fresh); - } + // W-3 — a reconcile throw out of either sleep transaction propagated past the + // success stamp and both callers (cron + webhook) swallowed it: neither + // success NOR failure was recorded, so `lastAttemptAt` sat quietly stale + // during a persistent write failure. Record a transient failure at the source + // AND rethrow — preserving the callers' existing warn-and-continue cohort + // semantics — so the write failure surfaces on the status card without + // touching either caller. The fetch-side `WithingsApiError` path already + // records + rethrows above, so it never reaches these blocks. + const recordSleepWriteFailure = async (err: unknown): Promise => { + await recordSyncFailure({ + userId, + integration: "withings", + kind: "transient", + message: `sleep write failed: ${ + err instanceof Error ? err.message : String(err) + }`, + }); + throw err; + }; - const verdict = await reconcileExternalMeasurement( - tx, - { - userId, - type: SLEEP_TYPE, - value: minutes, - unit: getUnitForType(SLEEP_TYPE), - measuredAt, - source: "WITHINGS", - sleepStage: stage, - externalId, - }, - { exactExternalMatch: "update" }, - ); - if (verdict.status === "failed") { - throw new MeasurementReconciliationError(verdict); + await prisma + .$transaction( + async (tx) => { + for (const segment of segments) { + const stage = mapWithingsSleepState(segment.state); + if (!stage) continue; + + const measuredAt = new Date(segment.enddate * 1000); + const durationSec = Math.max(0, segment.enddate - segment.startdate); + const minutes = Math.round(durationSec / 60); + const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`; + if (typeof segment.id === "number") { + const fresh = freshBySession.get(segment.id) ?? []; + fresh.push(externalId); + freshBySession.set(segment.id, fresh); + } + + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: SLEEP_TYPE, + value: minutes, + unit: getUnitForType(SLEEP_TYPE), + measuredAt, + source: "WITHINGS", + sleepStage: stage, + externalId, + }, + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); + } + if (verdict.status === "inserted") { + insertedSleepMeasuredAts.push(measuredAt); + } + touched.push({ type: SLEEP_TYPE, measuredAt }); + imported++; } - for (const dirty of verdict.dirtyIdentities ?? []) { - touched.push(dirty); - } - if (verdict.status === "inserted") { - insertedSleepMeasuredAts.push(measuredAt); - } - touched.push({ type: SLEEP_TYPE, measuredAt }); - imported++; - } - }, - { timeout: 60_000 }, - ); + }, + { timeout: 60_000 }, + ) + .catch(recordSleepWriteFailure); // Session-scoped sweep (mirrors Google Health's replace-by-window): for each // session this fetch re-produced, soft-delete any live WITHINGS @@ -489,38 +512,40 @@ export async function syncUserSleep( ); } - await prisma.$transaction( - async (tx) => { - for (const summary of summaries) { - const measuredAt = new Date(summary.enddate * 1000); - const sessionId = summary.id ?? "no-id"; - for (const vital of mapWithingsSleepSummary(summary.data)) { - const verdict = await reconcileExternalMeasurement( - tx, - { - userId, - type: vital.type, - value: vital.value, - unit: getUnitForType(vital.type), - measuredAt, - source: "WITHINGS", - externalId: `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`, - }, - { exactExternalMatch: "update" }, - ); - if (verdict.status === "failed") { - throw new MeasurementReconciliationError(verdict); + await prisma + .$transaction( + async (tx) => { + for (const summary of summaries) { + const measuredAt = new Date(summary.enddate * 1000); + const sessionId = summary.id ?? "no-id"; + for (const vital of mapWithingsSleepSummary(summary.data)) { + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: vital.type, + value: vital.value, + unit: getUnitForType(vital.type), + measuredAt, + source: "WITHINGS", + externalId: `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`, + }, + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); + } + touched.push({ type: vital.type, measuredAt }); + imported++; } - for (const dirty of verdict.dirtyIdentities ?? []) { - touched.push(dirty); - } - touched.push({ type: vital.type, measuredAt }); - imported++; } - } - }, - { timeout: 60_000 }, - ); + }, + { timeout: 60_000 }, + ) + .catch(recordSleepWriteFailure); // v1.4.39.1 — refresh the persistent rollup table for every distinct // (type, day) the sync touched. Sleep segments collapse heavily —