refactor(integrations): collapse the remaining duplication and drop a dead column

Derive the recovery source rank from the one priority ladder instead of a second
hand-kept map, re-point the WHOOP and Fitbit HTTP classifiers at the shared
status classifier, and lift the near-identical poll-cohort job handlers onto one
factory. Drop the Polar refresh-token column, which was only ever written null
since Polar tokens do not expire. Behaviour is unchanged across the recovery,
classifier, queue and route suites.

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;
 

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).

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),
       },
     });

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,
     },
   });

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,

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 <verb> error: <status>` 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");
 }

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);
+    }
+  });
 });

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<Record<MeasurementSource, number>> = {
-  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;
 }
 
 /**

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<NightscoutSyncPayload>[]) {
-  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,
+});

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<OuraSyncPayload>[]) {
-  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,
+});

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<PolarSyncPayload>[]) {
-  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,
+});

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<typeof getWorkerPrisma>) => Promise<string[]>;
+  /** Sync one user; returns the count of measurements imported. */
+  syncUser: (userId: string) => Promise<number>;
+}
+
+/**
+ * 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<PollCohortPayload>[]) {
+    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;
+      }
+    });
+  };
+}

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 <verb> error: <status>` 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");
 }