diff --git a/prisma/migrations/0160_v1170_new_measurement_sources/migration.sql b/prisma/migrations/0160_v1170_new_measurement_sources/migration.sql index 8931b5dfb..ce2d6c27c 100644 --- a/prisma/migrations/0160_v1170_new_measurement_sources/migration.sql +++ b/prisma/migrations/0160_v1170_new_measurement_sources/migration.sql @@ -40,8 +40,6 @@ ALTER TABLE "users" -- ── 3. users — Polar OAuth credentials (encrypted at app level) ──────── ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "polar_access_token_encrypted" TEXT; -ALTER TABLE "users" - ADD COLUMN IF NOT EXISTS "polar_refresh_token_encrypted" TEXT; ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "polar_user_id_encrypted" TEXT; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 38c126279..ba56aa388 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -126,7 +126,6 @@ model User { // product, so each self-hoster registers their own Polar app and the OAuth // tokens are stored encrypted. polarAccessTokenEncrypted String? @map("polar_access_token_encrypted") - polarRefreshTokenEncrypted String? @map("polar_refresh_token_encrypted") polarUserIdEncrypted String? @map("polar_user_id_encrypted") // v1.17.0 — Oura OAuth credentials (F4, per-user, encrypted at app level). diff --git a/src/app/api/polar/callback/route.ts b/src/app/api/polar/callback/route.ts index 10cfd08fc..34b084e45 100644 --- a/src/app/api/polar/callback/route.ts +++ b/src/app/api/polar/callback/route.ts @@ -94,8 +94,7 @@ export const GET = apiHandler(async (request: NextRequest) => { where: { id: userId }, data: { polarAccessTokenEncrypted: encrypt(tokens.access_token), - // Polar issues no refresh token; the column stays null. - polarRefreshTokenEncrypted: null, + // Polar issues no refresh token, so no refresh column is stored. polarUserIdEncrypted: encrypt(polarUserId), }, }); diff --git a/src/app/api/polar/disconnect/route.ts b/src/app/api/polar/disconnect/route.ts index f39a73692..07727ffa1 100644 --- a/src/app/api/polar/disconnect/route.ts +++ b/src/app/api/polar/disconnect/route.ts @@ -36,7 +36,6 @@ export const POST = apiHandler(async () => { where: { id: user.id }, data: { polarAccessTokenEncrypted: null, - polarRefreshTokenEncrypted: null, polarUserIdEncrypted: null, }, }); diff --git a/src/lib/doctor-report-data.ts b/src/lib/doctor-report-data.ts index 0aa635833..4b0aacdef 100644 --- a/src/lib/doctor-report-data.ts +++ b/src/lib/doctor-report-data.ts @@ -1082,6 +1082,9 @@ export async function collectDoctorReportData( stats: filteredStats, glucoseStats, glucoseRanges, + // Surfaced by the insights glucose views; the doctor-PDF / FHIR rendering of + // these clinical metrics is staged for a later release, so no report + // consumer reads this field yet — it is forward-populated, not dead. glucoseClinical, glucoseUnit: resolveGlucoseUnit(userProfile?.glucoseUnit ?? null), bmi: filteredBmi, diff --git a/src/lib/fitbit/response-classifier.ts b/src/lib/fitbit/response-classifier.ts index ca90b8290..46540168b 100644 --- a/src/lib/fitbit/response-classifier.ts +++ b/src/lib/fitbit/response-classifier.ts @@ -1,19 +1,17 @@ /** * Off-response classifier for Fitbit / Google Health API replies (v1.12.0). * - * Mirrors the WHOOP classifier (`src/lib/whoop/response-classifier.ts`) but - * speaks the Google Health wire. Like WHOOP, Google signals failure through the - * HTTP status code itself (no `status` field embedded in a 200 body), so this - * classifier is HTTP-status-driven. - * - * Classification buckets: + * Like WHOOP, Google signals failure through the HTTP status code itself (no + * `status` field embedded in a 200 body), so the classification is + * status-driven. The shared core lives in + * `src/lib/integrations/http-status-classifier.ts`; this module re-exports it + * under the Fitbit-specific names callers already use. * * - `success` → HTTP 2xx * - `transient` → retryable: HTTP 429 (rate-limited — honour * `Retry-After` for backoff at the call site), 5xx * upstream outages, 3xx CDN/network glitches, and an - * empty / off-spec body where we never reached the - * envelope. + * unknown status where we never reached the envelope. * - `reauth_required` → HTTP 401 / 403 — the access token is rejected or the * grant was revoked; the user must redo OAuth. (The * per-data-class 403 soft-skip lives in `sync.ts`'s @@ -24,99 +22,37 @@ * request or a contract mismatch that retrying without * operator intervention will not fix. * - * The classifier is conservative: anything it doesn't recognise defaults to - * `transient` so a single unknown response doesn't hard-disable the - * integration. The 3-strike admin-alert ladder in `recordSyncFailure` already - * catches the "keeps happening" case for recurring transients. - * * Unlike WHOOP, there is no `invalid_grant` token-body special-case: Google * signals a revoked refresh token with a 401 on the token endpoint, which the * status-driven classifier already buckets as `reauth_required`. */ -/** Outcome buckets for a Fitbit response. Same shape as WHOOP. */ -export type FitbitClassification = - | "success" - | "transient" - | "reauth_required" - | "persistent"; +import { + classifyHttpStatus, + classifyIntegrationError, + IntegrationApiError, + type ClassifiedHttpResponse, + type IntegrationClassification, +} from "@/lib/integrations/http-status-classifier"; -export interface ClassifiedFitbitResponse { - classification: FitbitClassification; - /** The HTTP status code that drove the verdict (undefined if none seen). */ - httpStatus: number | undefined; - /** Short human-readable label for logs / audit details. */ - reason: string; -} +/** Outcome buckets for a Fitbit response. */ +export type FitbitClassification = IntegrationClassification; -/** - * Classify a single Fitbit / Google Health API response by its HTTP status. - * - * @param httpStatus The HTTP status from fetch. - */ +export type ClassifiedFitbitResponse = ClassifiedHttpResponse; + +/** Classify a single Fitbit / Google Health API response by its HTTP status. */ export function classifyFitbitResponse( httpStatus: number, ): ClassifiedFitbitResponse { - if (httpStatus >= 200 && httpStatus < 300) { - return { classification: "success", httpStatus, reason: "ok" }; - } - // 429 is rate-limit — honour `Retry-After` for backoff at the call site, but - // the verdict is a plain transient so the next sync retries. - if (httpStatus === 429) { - return { classification: "transient", httpStatus, reason: "http_429" }; - } - // 401/403 — the bearer token is rejected or the grant revoked. The user has - // to reconnect; do not retry silently. (A per-data-class 403 soft-skip is - // applied in sync.ts before this verdict drives a connection-wide reauth.) - if (httpStatus === 401 || httpStatus === 403) { - return { - classification: "reauth_required", - httpStatus, - reason: `http_${httpStatus}`, - }; - } - // Any other 4xx is a malformed request / contract mismatch — persistent. - if (httpStatus >= 400 && httpStatus < 500) { - return { - classification: "persistent", - httpStatus, - reason: `http_${httpStatus}`, - }; - } - // 5xx upstream outage — retryable. - if (httpStatus >= 500 && httpStatus < 600) { - return { - classification: "transient", - httpStatus, - reason: `http_${httpStatus}`, - }; - } - // 3xx (a CDN / redirect glitch — safeFetch uses redirect:"manual", so a - // surfaced 3xx is unexpected) and everything else defaults to transient. - return { - classification: "transient", - httpStatus, - reason: `http_${httpStatus}_unknown`, - }; + return classifyHttpStatus(httpStatus); } /** - * Typed Error subclass carrying the classification verdict so downstream - * `try/catch` blocks can branch without re-parsing the message. Mirrors - * `WhoopApiError`. + * Typed error carrying the classification verdict so downstream `try/catch` + * blocks can branch without re-parsing the message. Extends the shared base + * (message shape `Fitbit error: ` is preserved). */ -export class FitbitApiError extends Error { - readonly classification: FitbitClassification; - readonly httpStatus: number | undefined; - readonly reason: string; - readonly verb: string; - /** - * The OAuth `error` code from a token-endpoint failure body (e.g. - * `invalid_grant`). Captured for the audit trail; the classifier does not - * branch on it (Google signals a revoked grant with a 401 status). - */ - readonly upstreamError: string | undefined; - +export class FitbitApiError extends IntegrationApiError { constructor(opts: { verb: string; classification: FitbitClassification; @@ -124,20 +60,8 @@ export class FitbitApiError extends Error { reason: string; upstreamError?: string; }) { - const statusLabel = - typeof opts.httpStatus === "number" ? opts.httpStatus : "?"; - const errSegment = opts.upstreamError ? ` - ${opts.upstreamError}` : ""; - // Cap the message at 1024 chars in the constructor (same bound as - // WhoopApiError) so a misbehaving upstream returning a multi-MB error body - // can't bloat an AuditLog row downstream. - const raw = `Fitbit ${opts.verb} error: ${statusLabel}${errSegment}`; - super(raw.slice(0, 1024)); + super({ vendor: "Fitbit", ...opts }); this.name = "FitbitApiError"; - this.verb = opts.verb; - this.classification = opts.classification; - this.httpStatus = opts.httpStatus; - this.reason = opts.reason; - this.upstreamError = opts.upstreamError; } } @@ -149,12 +73,5 @@ export class FitbitApiError extends Error { * for callers that lose the original prototype across a pg-boss retry. */ export function classifyFitbitError(err: unknown): FitbitClassification { - if (err instanceof FitbitApiError) return err.classification; - - const msg = err instanceof Error ? err.message : String(err); - const m = /Fitbit\s+\w+\s+error:\s*(\d+)/.exec(msg); - if (!m) return "transient"; - const status = Number.parseInt(m[1]!, 10); - if (!Number.isFinite(status)) return "transient"; - return classifyFitbitResponse(status).classification; + return classifyIntegrationError(err, "Fitbit"); } diff --git a/src/lib/insights/derived/__tests__/recovery-resolve.test.ts b/src/lib/insights/derived/__tests__/recovery-resolve.test.ts index b393ee458..df9487738 100644 --- a/src/lib/insights/derived/__tests__/recovery-resolve.test.ts +++ b/src/lib/insights/derived/__tests__/recovery-resolve.test.ts @@ -3,6 +3,7 @@ import { resolveCanonicalRecovery, type RecoveryRow, } from "../recovery-resolve"; +import { DEFAULT_SOURCE_PRIORITY } from "@/lib/validations/source-priority"; function row(iso: string, value: number, source: string): RecoveryRow { return { value, measuredAt: new Date(iso), source: source as never }; @@ -116,4 +117,26 @@ describe("resolveCanonicalRecovery", () => { it("returns an empty list for no rows", () => { expect(resolveCanonicalRecovery([])).toEqual([]); }); + + // The per-source authority is derived from `DEFAULT_SOURCE_PRIORITY.recovery` + // (one ordered ladder), not a second hardcoded rank map. For every adjacent + // pair on that ladder the earlier source must win the same night, proving the + // resolution order tracks the ladder rather than a copy of it. + it("resolves in the order of DEFAULT_SOURCE_PRIORITY.recovery", () => { + const ladder = DEFAULT_SOURCE_PRIORITY.recovery; + for (let i = 0; i < ladder.length - 1; i++) { + const higher = ladder[i]; + const lower = ladder[i + 1]; + // COMPUTED carries the day-that-ended stamp; everything else stamps the + // wake morning. Stamp both for the SAME wake day so they collide. + const stampFor = (src: string) => + src === "COMPUTED" ? "2026-06-01T12:00:00Z" : "2026-06-02T06:00:00Z"; + const resolved = resolveCanonicalRecovery([ + row(stampFor(lower), 10, lower), + row(stampFor(higher), 90, higher), + ]); + expect(resolved).toHaveLength(1); + expect(resolved[0].source).toBe(higher); + } + }); }); diff --git a/src/lib/insights/derived/recovery-resolve.ts b/src/lib/insights/derived/recovery-resolve.ts index 4f8c8cfed..b87298301 100644 --- a/src/lib/insights/derived/recovery-resolve.ts +++ b/src/lib/insights/derived/recovery-resolve.ts @@ -41,6 +41,7 @@ import type { MeasurementSource } from "@/generated/prisma/client"; import { dayKeyForUserTz } from "@/lib/measurements/consolidation-tz"; import { resolveUserTimezone } from "@/lib/measurements/consolidation-base"; +import { DEFAULT_SOURCE_PRIORITY } from "@/lib/validations/source-priority"; const MS_PER_DAY = 24 * 60 * 60 * 1000; @@ -53,22 +54,19 @@ export interface RecoveryRow { /** * The per-source preference for `RECOVERY_SCORE`: a native device row outranks - * the COMPUTED proxy for the same day. Lower number = higher authority. The - * ladder mirrors `DEFAULT_SOURCE_PRIORITY.recovery` - * (`WHOOP > OURA > POLAR > COMPUTED`): WHOOP's all-night strap leads, then the - * Oura ring readiness and the Polar Nightly Recharge band, then the computed - * proxy as the fallback for users without a wearable. A source not listed falls - * below all of them. + * the COMPUTED proxy for the same day. Authority is read straight off + * `DEFAULT_SOURCE_PRIORITY.recovery` (`WHOOP > OURA > POLAR > COMPUTED`) — that + * ordered ladder is the single source of truth, so adding a recovery source + * updates only the priority list and this resolver inherits the new rank for + * free. WHOOP's all-night strap leads, then the Oura ring readiness and the + * Polar Nightly Recharge band, then the computed proxy as the fallback for + * users without a wearable. A source not on the ladder falls below all of them. */ -const RECOVERY_SOURCE_RANK: Partial> = { - WHOOP: 0, - OURA: 1, - POLAR: 2, - COMPUTED: 3, -}; +const RECOVERY_LADDER = DEFAULT_SOURCE_PRIORITY.recovery; function rankOf(source: MeasurementSource): number { - return RECOVERY_SOURCE_RANK[source] ?? Number.MAX_SAFE_INTEGER; + const idx = RECOVERY_LADDER.indexOf(source); + return idx === -1 ? Number.MAX_SAFE_INTEGER : idx; } /** diff --git a/src/lib/jobs/reminder/nightscout-sync.ts b/src/lib/jobs/reminder/nightscout-sync.ts index a05d68d14..1a97f24b1 100644 --- a/src/lib/jobs/reminder/nightscout-sync.ts +++ b/src/lib/jobs/reminder/nightscout-sync.ts @@ -5,66 +5,24 @@ * hourly cron tick carries no `userId` and the handler iterates every user with * a configured instance, re-syncing each via `syncUserNightscout`. One user's * unreachable instance never starves the rest of the cohort — a per-user error - * is warned and the pass continues. - * - * Extracted from reminder-worker.ts, which owns the queue name, cron schedule, - * and boss.work registration. + * is warned and the pass continues. The shared control flow lives in + * `./poll-cohort`; reminder-worker.ts owns the queue name, cron schedule, and + * boss.work registration. */ -import { type Job } from "pg-boss"; - -import { recordError } from "@/lib/jobs/worker-status"; -import { withBackgroundEvent } from "@/lib/logging/background"; import { syncUserNightscout } from "@/lib/nightscout/sync"; -import { getWorkerPrisma } from "./shared"; +import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort"; -export interface NightscoutSyncPayload { - userId?: string; -} +export type NightscoutSyncPayload = PollCohortPayload; -export async function handleNightscoutSync(jobs: Job[]) { - await withBackgroundEvent("job.nightscout_sync", async (evt) => { - const prisma = getWorkerPrisma(); - try { - const targets: Array<{ userId: string }> = []; - for (const job of jobs) { - if (job.data?.userId) targets.push({ userId: job.data.userId }); - } - if (targets.length === 0) { - // Cron tick — re-walk every configured instance. The URL column is - // the connect marker (a public instance stores no token). - const users = await prisma.user.findMany({ - where: { nightscoutUrlEncrypted: { not: null } }, - select: { id: true }, - }); - targets.push(...users.map((u) => ({ userId: u.id }))); - } - if (targets.length === 0) return; - - let usersSynced = 0; - let measurementsImported = 0; - for (const { userId } of targets) { - try { - measurementsImported += await syncUserNightscout(userId); - usersSynced++; - } catch (err) { - evt.addWarning( - `job.nightscout_sync failed for user ${userId}: ${err}`, - ); - } - } - - evt.setBackground({ - task_name: "job.nightscout_sync", - result: { - users_synced: usersSynced, - total: targets.length, - measurements_imported: measurementsImported, - }, - }); - } catch (err) { - evt.setError(err); - recordError(); - throw err; - } - }); -} +export const handleNightscoutSync = makePollCohortHandler({ + taskName: "job.nightscout_sync", + findCohort: async (prisma) => { + // The URL column is the connect marker (a public instance stores no token). + const users = await prisma.user.findMany({ + where: { nightscoutUrlEncrypted: { not: null } }, + select: { id: true }, + }); + return users.map((u) => u.id); + }, + syncUser: syncUserNightscout, +}); diff --git a/src/lib/jobs/reminder/oura-sync.ts b/src/lib/jobs/reminder/oura-sync.ts index b722b89ad..5ca264c4e 100644 --- a/src/lib/jobs/reminder/oura-sync.ts +++ b/src/lib/jobs/reminder/oura-sync.ts @@ -3,60 +3,23 @@ * * Poll-only: one hourly cron tick (no `userId`) iterates every user with a * stored Oura token and re-syncs each via `syncUserOura`. One user's revoked - * grant / outage is warned and the pass continues. The queue name, cron - * schedule, and boss.work registration live in reminder-worker.ts. + * grant / outage is warned and the pass continues. The shared control flow + * lives in `./poll-cohort`; the queue name, cron schedule, and boss.work + * registration live in reminder-worker.ts. */ -import { type Job } from "pg-boss"; - -import { recordError } from "@/lib/jobs/worker-status"; -import { withBackgroundEvent } from "@/lib/logging/background"; import { syncUserOura } from "@/lib/oura/sync"; -import { getWorkerPrisma } from "./shared"; +import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort"; -export interface OuraSyncPayload { - userId?: string; -} +export type OuraSyncPayload = PollCohortPayload; -export async function handleOuraSync(jobs: Job[]) { - await withBackgroundEvent("job.oura_sync", async (evt) => { - const prisma = getWorkerPrisma(); - try { - const targets: Array<{ userId: string }> = []; - for (const job of jobs) { - if (job.data?.userId) targets.push({ userId: job.data.userId }); - } - if (targets.length === 0) { - const users = await prisma.user.findMany({ - where: { ouraAccessTokenEncrypted: { not: null } }, - select: { id: true }, - }); - targets.push(...users.map((u) => ({ userId: u.id }))); - } - if (targets.length === 0) return; - - let usersSynced = 0; - let measurementsImported = 0; - for (const { userId } of targets) { - try { - measurementsImported += await syncUserOura(userId); - usersSynced++; - } catch (err) { - evt.addWarning(`job.oura_sync failed for user ${userId}: ${err}`); - } - } - - evt.setBackground({ - task_name: "job.oura_sync", - result: { - users_synced: usersSynced, - total: targets.length, - measurements_imported: measurementsImported, - }, - }); - } catch (err) { - evt.setError(err); - recordError(); - throw err; - } - }); -} +export const handleOuraSync = makePollCohortHandler({ + taskName: "job.oura_sync", + findCohort: async (prisma) => { + const users = await prisma.user.findMany({ + where: { ouraAccessTokenEncrypted: { not: null } }, + select: { id: true }, + }); + return users.map((u) => u.id); + }, + syncUser: syncUserOura, +}); diff --git a/src/lib/jobs/reminder/polar-sync.ts b/src/lib/jobs/reminder/polar-sync.ts index 332bce2c8..2ae9c78df 100644 --- a/src/lib/jobs/reminder/polar-sync.ts +++ b/src/lib/jobs/reminder/polar-sync.ts @@ -5,62 +5,24 @@ * poll model (matching Nightscout). One hourly cron tick carries no `userId` * and the handler iterates every user with a stored Polar token, re-syncing * each via `syncUserPolar`. One user's revoked grant / outage is warned and the - * pass continues. + * pass continues. The shared control flow lives in `./poll-cohort`. * * The queue name, cron schedule, and boss.work registration live in * reminder-worker.ts. */ -import { type Job } from "pg-boss"; - -import { recordError } from "@/lib/jobs/worker-status"; -import { withBackgroundEvent } from "@/lib/logging/background"; import { syncUserPolar } from "@/lib/polar/sync"; -import { getWorkerPrisma } from "./shared"; +import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort"; -export interface PolarSyncPayload { - userId?: string; -} +export type PolarSyncPayload = PollCohortPayload; -export async function handlePolarSync(jobs: Job[]) { - await withBackgroundEvent("job.polar_sync", async (evt) => { - const prisma = getWorkerPrisma(); - try { - const targets: Array<{ userId: string }> = []; - for (const job of jobs) { - if (job.data?.userId) targets.push({ userId: job.data.userId }); - } - if (targets.length === 0) { - const users = await prisma.user.findMany({ - where: { polarAccessTokenEncrypted: { not: null } }, - select: { id: true }, - }); - targets.push(...users.map((u) => ({ userId: u.id }))); - } - if (targets.length === 0) return; - - let usersSynced = 0; - let measurementsImported = 0; - for (const { userId } of targets) { - try { - measurementsImported += await syncUserPolar(userId); - usersSynced++; - } catch (err) { - evt.addWarning(`job.polar_sync failed for user ${userId}: ${err}`); - } - } - - evt.setBackground({ - task_name: "job.polar_sync", - result: { - users_synced: usersSynced, - total: targets.length, - measurements_imported: measurementsImported, - }, - }); - } catch (err) { - evt.setError(err); - recordError(); - throw err; - } - }); -} +export const handlePolarSync = makePollCohortHandler({ + taskName: "job.polar_sync", + findCohort: async (prisma) => { + const users = await prisma.user.findMany({ + where: { polarAccessTokenEncrypted: { not: null } }, + select: { id: true }, + }); + return users.map((u) => u.id); + }, + syncUser: syncUserPolar, +}); diff --git a/src/lib/jobs/reminder/poll-cohort.ts b/src/lib/jobs/reminder/poll-cohort.ts new file mode 100644 index 000000000..e07bdcc18 --- /dev/null +++ b/src/lib/jobs/reminder/poll-cohort.ts @@ -0,0 +1,94 @@ +/** + * v1.17.0 — shared poll-cohort job-handler factory. + * + * The Polar, Oura and Nightscout sync handlers were byte-identical wrappers + * differing only in three things: the wide-event task name, the cohort + * discovery query (which credential column marks a connected user), and the + * per-user `syncUser*` callback. This factory lifts the shared control flow — + * targets from the job payloads or, on a `userId`-less cron tick, from the + * cohort query; a sequential pass with per-user error isolation; and the + * `users_synced` / `total` / `measurements_imported` background result. + * + * Each provider file collapses to a small config calling `makePollCohortHandler`. + * The queue name, cron schedule, and boss.work registration stay in + * reminder-worker.ts. + * + * Fitbit is intentionally NOT built on this factory: it fans the cohort out + * with bounded concurrency via `runFitbitPollCohort`, a different control flow + * from this sequential pass. + */ +import { type Job } from "pg-boss"; + +import { recordError } from "@/lib/jobs/worker-status"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { getWorkerPrisma } from "./shared"; + +/** Every poll-cohort handler accepts an optional single-user payload. */ +export interface PollCohortPayload { + userId?: string; +} + +export interface PollCohortConfig { + /** Wide-event task name, e.g. `job.polar_sync`. */ + taskName: string; + /** + * Cohort discovery for a `userId`-less cron tick: returns every connected + * user's id. Receives the worker Prisma client. + */ + findCohort: (prisma: ReturnType) => Promise; + /** Sync one user; returns the count of measurements imported. */ + syncUser: (userId: string) => Promise; +} + +/** + * Build a pg-boss batch handler for a poll-only integration. The returned + * handler resolves its targets from the batch payloads (single-user enqueue) or + * the cohort query (cron tick), then re-syncs each sequentially with per-user + * error isolation so one revoked grant / outage never starves the cohort. + */ +export function makePollCohortHandler({ + taskName, + findCohort, + syncUser, +}: PollCohortConfig) { + return async function handlePollCohort(jobs: Job[]) { + await withBackgroundEvent(taskName, async (evt) => { + const prisma = getWorkerPrisma(); + try { + const targets: Array<{ userId: string }> = []; + for (const job of jobs) { + if (job.data?.userId) targets.push({ userId: job.data.userId }); + } + if (targets.length === 0) { + const ids = await findCohort(prisma); + targets.push(...ids.map((userId) => ({ userId }))); + } + if (targets.length === 0) return; + + let usersSynced = 0; + let measurementsImported = 0; + for (const { userId } of targets) { + try { + measurementsImported += await syncUser(userId); + usersSynced++; + } catch (err) { + evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); + } + } + + evt.setBackground({ + task_name: taskName, + result: { + users_synced: usersSynced, + total: targets.length, + measurements_imported: measurementsImported, + }, + }); + } catch (err) { + evt.setError(err); + recordError(); + throw err; + } + }); + }; +} diff --git a/src/lib/whoop/response-classifier.ts b/src/lib/whoop/response-classifier.ts index b6b5538ce..792815d83 100644 --- a/src/lib/whoop/response-classifier.ts +++ b/src/lib/whoop/response-classifier.ts @@ -1,112 +1,50 @@ /** * Off-response classifier for WHOOP API replies. * - * Mirrors the Withings classifier (`src/lib/withings/response-classifier.ts`) - * but speaks the WHOOP wire instead. WHOOP, unlike Withings, signals failure - * through the HTTP status code itself (no `status` field embedded in a 200 - * body), so this classifier is HTTP-status-driven. - * - * Classification buckets: + * WHOOP signals failure through the HTTP status code itself (no `status` field + * embedded in a 200 body), so the classification is status-driven. The shared + * core lives in `src/lib/integrations/http-status-classifier.ts`; this module + * re-exports it under the WHOOP-specific names callers already use. * * - `success` → HTTP 2xx * - `transient` → retryable: HTTP 429 (rate-limited — WHOOP caps at * 100 req/min + 10 000 req/day per app), 5xx upstream - * outages, 3xx CDN/network glitches, and an empty / - * off-spec body where we never reached the envelope. + * outages, 3xx CDN/network glitches, and an unknown + * status where we never reached the envelope. * - `reauth_required` → HTTP 401 / 403 — the access token is rejected or the * grant was revoked; the user must redo OAuth. * - `persistent` → any other 4xx (400, 404, 422, …) — a malformed * request or a contract mismatch that retrying without * operator intervention will not fix. - * - * The classifier is conservative: anything it doesn't recognise defaults to - * `transient` so a single unknown response doesn't hard-disable the - * integration. The 3-strike admin-alert ladder in `recordSyncFailure` - * already catches the "keeps happening" case for recurring transients. */ -/** Outcome buckets for a WHOOP response. Same shape as Withings. */ -export type WhoopClassification = - | "success" - | "transient" - | "reauth_required" - | "persistent"; +import { + classifyHttpStatus, + classifyIntegrationError, + IntegrationApiError, + isOAuthInvalidGrant, + type ClassifiedHttpResponse, + type IntegrationClassification, +} from "@/lib/integrations/http-status-classifier"; -export interface ClassifiedWhoopResponse { - classification: WhoopClassification; - /** The HTTP status code that drove the verdict (undefined if none seen). */ - httpStatus: number | undefined; - /** Short human-readable label for logs / audit details. */ - reason: string; -} +/** Outcome buckets for a WHOOP response. */ +export type WhoopClassification = IntegrationClassification; -/** - * Classify a single WHOOP API response by its HTTP status code. - * - * @param httpStatus The HTTP status from fetch. - */ +export type ClassifiedWhoopResponse = ClassifiedHttpResponse; + +/** Classify a single WHOOP API response by its HTTP status code. */ export function classifyWhoopResponse( httpStatus: number, ): ClassifiedWhoopResponse { - if (httpStatus >= 200 && httpStatus < 300) { - return { classification: "success", httpStatus, reason: "ok" }; - } - // 429 is rate-limit — honour `X-RateLimit-*` for backoff at the call site, - // but the verdict is a plain transient so the next sync retries. - if (httpStatus === 429) { - return { classification: "transient", httpStatus, reason: "http_429" }; - } - // 401/403 — the bearer token is rejected or the grant revoked. The user has - // to reconnect; do not retry silently. - if (httpStatus === 401 || httpStatus === 403) { - return { - classification: "reauth_required", - httpStatus, - reason: `http_${httpStatus}`, - }; - } - // Any other 4xx is a malformed request / contract mismatch — persistent. - if (httpStatus >= 400 && httpStatus < 500) { - return { - classification: "persistent", - httpStatus, - reason: `http_${httpStatus}`, - }; - } - // 5xx upstream outage — retryable. - if (httpStatus >= 500 && httpStatus < 600) { - return { - classification: "transient", - httpStatus, - reason: `http_${httpStatus}`, - }; - } - // 3xx (a CDN / redirect glitch — safeFetch uses redirect:"manual", so a - // surfaced 3xx is unexpected) and everything else defaults to transient. - return { - classification: "transient", - httpStatus, - reason: `http_${httpStatus}_unknown`, - }; + return classifyHttpStatus(httpStatus); } /** - * Typed Error subclass carrying the classification verdict so downstream - * `try/catch` blocks can branch without re-parsing the message. Mirrors - * `WithingsApiError`. + * Typed error carrying the classification verdict so downstream `try/catch` + * blocks can branch without re-parsing the message. Extends the shared base + * (message shape `WHOOP error: ` is preserved). */ -export class WhoopApiError extends Error { - readonly classification: WhoopClassification; - readonly httpStatus: number | undefined; - readonly reason: string; - readonly verb: string; - /** - * The OAuth `error` code from a token-endpoint failure body (e.g. - * `invalid_grant` when a refresh token is revoked). Undefined for the - * collection endpoints, which signal failure through the status code alone. - */ - readonly upstreamError: string | undefined; - +export class WhoopApiError extends IntegrationApiError { constructor(opts: { verb: string; classification: WhoopClassification; @@ -114,20 +52,8 @@ export class WhoopApiError extends Error { reason: string; upstreamError?: string; }) { - const statusLabel = - typeof opts.httpStatus === "number" ? opts.httpStatus : "?"; - const errSegment = opts.upstreamError ? ` - ${opts.upstreamError}` : ""; - // Cap the message at 1024 chars in the constructor (same bound as - // WithingsApiError) so a misbehaving upstream returning a multi-MB error - // body can't bloat an AuditLog row downstream. - const raw = `WHOOP ${opts.verb} error: ${statusLabel}${errSegment}`; - super(raw.slice(0, 1024)); + super({ vendor: "WHOOP", ...opts }); this.name = "WhoopApiError"; - this.verb = opts.verb; - this.classification = opts.classification; - this.httpStatus = opts.httpStatus; - this.reason = opts.reason; - this.upstreamError = opts.upstreamError; } } @@ -140,19 +66,9 @@ export class WhoopApiError extends Error { * surface the reauth prompt. Scoped tightly to a 400 carrying `invalid_grant` * so other 400s (a malformed request, a bad client_secret → `invalid_client`) * stay persistent. - * - * Robust to a lost prototype across a pg-boss retry: it reads the dedicated - * `upstreamError` field when present, else falls back to the - * `"... error: 400 - invalid_grant"` message segment. */ export function isInvalidGrant(err: unknown): boolean { - if (err instanceof WhoopApiError) { - if (err.httpStatus !== 400) return false; - if (err.upstreamError === "invalid_grant") return true; - return /\binvalid_grant\b/.test(err.message); - } - const msg = err instanceof Error ? err.message : String(err); - return /WHOOP\s+\w+\s+error:\s*400\b[\s\S]*\binvalid_grant\b/.test(msg); + return isOAuthInvalidGrant(err, "WHOOP"); } /** @@ -168,13 +84,5 @@ export function isInvalidGrant(err: unknown): boolean { */ export function classifyWhoopError(err: unknown): WhoopClassification { if (isInvalidGrant(err)) return "reauth_required"; - - if (err instanceof WhoopApiError) return err.classification; - - const msg = err instanceof Error ? err.message : String(err); - const m = /WHOOP\s+\w+\s+error:\s*(\d+)/.exec(msg); - if (!m) return "transient"; - const status = Number.parseInt(m[1]!, 10); - if (!Number.isFinite(status)) return "transient"; - return classifyWhoopResponse(status).classification; + return classifyIntegrationError(err, "WHOOP"); }