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