Extract poll-cohort job handler factory

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