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/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/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/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/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");