diff --git a/src/__tests__/backfill-preserves-updated-at-guard.test.ts b/src/__tests__/backfill-preserves-updated-at-guard.test.ts index 39fe550ab..7531d6bd0 100644 --- a/src/__tests__/backfill-preserves-updated-at-guard.test.ts +++ b/src/__tests__/backfill-preserves-updated-at-guard.test.ts @@ -25,7 +25,7 @@ * true, only that the set did not change without someone editing this file. */ import { describe, it, expect } from "vitest"; -import { readFileSync, readdirSync } from "node:fs"; +import { existsSync, readFileSync, readdirSync } from "node:fs"; import { join } from "node:path"; import { walkSourceFiles } from "./helpers/source-files"; @@ -84,17 +84,25 @@ function backfillFiles(): string[] { const jobs = readdirSync(join(SRC, "lib/jobs")) .filter((f) => /-backfill\.ts$/.test(f)) .map((f) => `src/lib/jobs/${f}`); + // A provider that carries its own jobs folder (`lib//jobs/`). + const providerJobs = readdirSync(join(SRC, "lib")).flatMap((dir) => { + const jobsDir = join(SRC, "lib", dir, "jobs"); + if (!existsSync(jobsDir)) return []; + return readdirSync(jobsDir) + .filter((f) => /(?:^|-)backfill\.ts$/.test(f)) + .map((f) => `src/lib/${dir}/jobs/${f}`); + }); const scripts = readdirSync(join(ROOT, "scripts")) .filter((f) => /^backfill-.*\.ts$/.test(f)) .map((f) => `scripts/${f}`); - return [...jobs, ...scripts].sort(); + return [...jobs, ...providerJobs, ...scripts].sort(); } /** * Backfill writes that stamp `updatedAt` on purpose, keyed `file > model`. */ const BACKFILL_STAMPS: Record = { - "src/lib/jobs/fitbit-backfill.ts > fitbitConnection": + "src/lib/fitbit/jobs/backfill.ts > fitbitConnection": "the job's own progress on the connection row; not a listed record", "src/lib/jobs/google-health-backfill.ts > googleHealthConnection": "the job's own progress on the connection row; not a listed record", diff --git a/src/__tests__/baselines/test-source-reads.json b/src/__tests__/baselines/test-source-reads.json index 309eef2f2..1a976c5f6 100644 --- a/src/__tests__/baselines/test-source-reads.json +++ b/src/__tests__/baselines/test-source-reads.json @@ -320,7 +320,7 @@ "lib/jobs/__tests__/data-arrival-provider-isolation.test.ts": 1, "lib/jobs/__tests__/data-arrival-queue.test.ts": 1, "lib/jobs/__tests__/data-backup-send-policy.test.ts": 1, - "lib/jobs/__tests__/fitbit-queues.test.ts": 1, + "lib/fitbit/jobs/__tests__/queues.test.ts": 1, "lib/jobs/__tests__/geo-backfill.test.ts": 1, "lib/jobs/__tests__/geolite2-fetch-queue.test.ts": 1, "lib/jobs/__tests__/google-health-queues.test.ts": 1, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index ad663138c..46f890774 100644 --- a/src/__tests__/baselines/version-stamp-comments.json +++ b/src/__tests__/baselines/version-stamp-comments.json @@ -2292,7 +2292,7 @@ "lib/jobs/__tests__/coach-reminder-sweep.test.ts": 1, "lib/jobs/__tests__/cycle-reminder.test.ts": 2, "lib/jobs/__tests__/feedback-aggregator.test.ts": 2, - "lib/jobs/__tests__/fitbit-backfill.test.ts": 1, + "lib/fitbit/jobs/__tests__/backfill.test.ts": 1, "lib/jobs/__tests__/geo-backfill.test.ts": 1, "lib/jobs/__tests__/google-health-sleep-repair.test.ts": 1, "lib/jobs/__tests__/hero-line-json-envelope.test.ts": 1, @@ -2336,8 +2336,8 @@ "lib/jobs/environment-air-quality-history.ts": 1, "lib/jobs/environment-fetch.ts": 2, "lib/jobs/feedback-aggregator.ts": 6, - "lib/jobs/fitbit-backfill.ts": 1, - "lib/jobs/fitbit-oauth-state-cleanup.ts": 1, + "lib/fitbit/jobs/backfill.ts": 1, + "lib/fitbit/jobs/oauth-state-cleanup.ts": 1, "lib/jobs/free-text-encryption-backfill.ts": 8, "lib/jobs/geo-backfill.ts": 4, "lib/jobs/google-health-backfill.ts": 1, @@ -2374,7 +2374,7 @@ "lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts": 1, "lib/jobs/reminder/__tests__/queue-retention.test.ts": 1, "lib/jobs/reminder/cleanup-handlers.ts": 7, - "lib/jobs/reminder/fitbit-sync.ts": 2, + "lib/fitbit/jobs/sync-handlers.ts": 2, "lib/jobs/reminder/google-health-sync.ts": 1, "lib/jobs/reminder/insights-handlers.ts": 3, "lib/jobs/reminder/medication-maintenance.ts": 3, diff --git a/src/lib/jobs/__tests__/fitbit-backfill.test.ts b/src/lib/fitbit/jobs/__tests__/backfill.test.ts similarity index 99% rename from src/lib/jobs/__tests__/fitbit-backfill.test.ts rename to src/lib/fitbit/jobs/__tests__/backfill.test.ts index fb1720bfa..6f17a4431 100644 --- a/src/lib/jobs/__tests__/fitbit-backfill.test.ts +++ b/src/lib/fitbit/jobs/__tests__/backfill.test.ts @@ -42,7 +42,7 @@ vi.mock("@/lib/logging/context", () => ({ import { enqueueBootTimeFitbitBackfill, runFitbitBackfillForUser, -} from "../fitbit-backfill"; +} from "@/lib/fitbit/jobs/backfill"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; beforeEach(() => { diff --git a/src/lib/jobs/__tests__/fitbit-queues.test.ts b/src/lib/fitbit/jobs/__tests__/queues.test.ts similarity index 97% rename from src/lib/jobs/__tests__/fitbit-queues.test.ts rename to src/lib/fitbit/jobs/__tests__/queues.test.ts index 988c8fcf8..6f6db73f9 100644 --- a/src/lib/jobs/__tests__/fitbit-queues.test.ts +++ b/src/lib/fitbit/jobs/__tests__/queues.test.ts @@ -29,7 +29,7 @@ import { INTEGRATION_BACKFILL_ADMISSION_QUEUE } from "@/lib/jobs/integration-bac import { handleFitbitOAuthStateCleanup, handleFitbitSync, -} from "@/lib/jobs/reminder/fitbit-sync"; +} from "@/lib/fitbit/jobs/sync-handlers"; import { cronIsTheRetry } from "@/lib/jobs/reminder/registrar-shared"; import { discoveryOf, @@ -40,7 +40,7 @@ import { } from "@/lib/jobs/registry/__tests__/registry-helpers"; const handlerSource = readFileSync( - join(__dirname, "..", "reminder", "fitbit-sync.ts"), + join(__dirname, "..", "sync-handlers.ts"), "utf8", ); diff --git a/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts b/src/lib/fitbit/jobs/__tests__/sleep-repair.test.ts similarity index 99% rename from src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts rename to src/lib/fitbit/jobs/__tests__/sleep-repair.test.ts index 23922aee7..374c6fe43 100644 --- a/src/lib/jobs/__tests__/fitbit-sleep-repair.test.ts +++ b/src/lib/fitbit/jobs/__tests__/sleep-repair.test.ts @@ -76,7 +76,7 @@ import { import { enqueueBootTimeFitbitSleepRepair, runFitbitSleepRepairForUser, -} from "../fitbit-sleep-repair"; +} from "@/lib/fitbit/jobs/sleep-repair"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; beforeEach(() => { diff --git a/src/lib/jobs/__tests__/fitbit-sync-retired.test.ts b/src/lib/fitbit/jobs/__tests__/sync-retired.test.ts similarity index 95% rename from src/lib/jobs/__tests__/fitbit-sync-retired.test.ts rename to src/lib/fitbit/jobs/__tests__/sync-retired.test.ts index 445930b16..7d5e5ef7f 100644 --- a/src/lib/jobs/__tests__/fitbit-sync-retired.test.ts +++ b/src/lib/fitbit/jobs/__tests__/sync-retired.test.ts @@ -21,7 +21,7 @@ vi.mock("@/lib/jobs/reminder-satisfy", () => ({ enqueueReminderSatisfy: vi.fn(async () => {}), })); -import { handleFitbitSync } from "../reminder/fitbit-sync"; +import { handleFitbitSync } from "@/lib/fitbit/jobs/sync-handlers"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; describe("handleFitbitSync", () => { diff --git a/src/lib/jobs/fitbit-backfill.ts b/src/lib/fitbit/jobs/backfill.ts similarity index 98% rename from src/lib/jobs/fitbit-backfill.ts rename to src/lib/fitbit/jobs/backfill.ts index 02f84fa21..69d81a05f 100644 --- a/src/lib/jobs/fitbit-backfill.ts +++ b/src/lib/fitbit/jobs/backfill.ts @@ -21,7 +21,7 @@ import { } from "@/lib/integrations/page-walk"; import { syncUserFitbit } from "@/lib/fitbit/sync"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; -import { FITBIT_BACKFILL_QUEUE } from "./../fitbit/jobs/queues"; +import { FITBIT_BACKFILL_QUEUE } from "./queues"; export interface FitbitBackfillPayload { userId: string; diff --git a/src/lib/fitbit/jobs/definitions.ts b/src/lib/fitbit/jobs/definitions.ts index a4308c82d..c01d6f31e 100644 --- a/src/lib/fitbit/jobs/definitions.ts +++ b/src/lib/fitbit/jobs/definitions.ts @@ -33,8 +33,7 @@ export const FITBIT_JOBS: readonly JobDefinition[] = [ queue: FITBIT_SYNC_QUEUE, cron: FITBIT_SYNC_CRON, work: { localConcurrency: 1 }, - load: async () => - (await import("@/lib/jobs/reminder/fitbit-sync")).handleFitbitSync, + load: async () => (await import("./sync-handlers")).handleFitbitSync, }), // The source queue keeps the per-connection singleton and forwards // admitted work to the shared drain. @@ -72,8 +71,7 @@ export const FITBIT_JOBS: readonly JobDefinition[] = [ scheduleOptions: cronIsTheRetry, work: { localConcurrency: 1 }, load: async () => - (await import("@/lib/jobs/reminder/fitbit-sync")) - .handleFitbitOAuthStateCleanup, + (await import("./sync-handlers")).handleFitbitOAuthStateCleanup, }), ]; @@ -82,16 +80,14 @@ export const FITBIT_BOOT_DISCOVERY: readonly BootDiscovery[] = [ { name: "fitbit-backfill", run: async () => - ( - await import("@/lib/jobs/fitbit-backfill") - ).enqueueBootTimeFitbitBackfill(bootStaggerSecondsFor("fitbit-backfill")), + (await import("./backfill")).enqueueBootTimeFitbitBackfill( + bootStaggerSecondsFor("fitbit-backfill"), + ), }, { name: "fitbit-sleep-repair", run: async () => - ( - await import("@/lib/jobs/fitbit-sleep-repair") - ).enqueueBootTimeFitbitSleepRepair( + (await import("./sleep-repair")).enqueueBootTimeFitbitSleepRepair( bootStaggerSecondsFor("fitbit-sleep-repair"), ), }, diff --git a/src/lib/jobs/fitbit-oauth-state-cleanup.ts b/src/lib/fitbit/jobs/oauth-state-cleanup.ts similarity index 100% rename from src/lib/jobs/fitbit-oauth-state-cleanup.ts rename to src/lib/fitbit/jobs/oauth-state-cleanup.ts diff --git a/src/lib/jobs/fitbit-sleep-repair.ts b/src/lib/fitbit/jobs/sleep-repair.ts similarity index 99% rename from src/lib/jobs/fitbit-sleep-repair.ts rename to src/lib/fitbit/jobs/sleep-repair.ts index f79ebb8f2..f4919a803 100644 --- a/src/lib/jobs/fitbit-sleep-repair.ts +++ b/src/lib/fitbit/jobs/sleep-repair.ts @@ -46,7 +46,7 @@ import { } from "@/lib/fitbit/sync-core"; import { syncUserSleep } from "@/lib/fitbit/sync-sleep"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; -import { FITBIT_SLEEP_REPAIR_QUEUE } from "./../fitbit/jobs/queues"; +import { FITBIT_SLEEP_REPAIR_QUEUE } from "./queues"; export interface FitbitSleepRepairPayload { userId: string; diff --git a/src/lib/jobs/reminder/fitbit-sync.ts b/src/lib/fitbit/jobs/sync-handlers.ts similarity index 97% rename from src/lib/jobs/reminder/fitbit-sync.ts rename to src/lib/fitbit/jobs/sync-handlers.ts index 9a560eb38..3b9ea163d 100644 --- a/src/lib/jobs/reminder/fitbit-sync.ts +++ b/src/lib/fitbit/jobs/sync-handlers.ts @@ -11,8 +11,8 @@ import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { runFitbitPollCohort } from "@/lib/fitbit/sync"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; -import { cleanupExpiredFitbitOAuthStates } from "@/lib/jobs/fitbit-oauth-state-cleanup"; -import { getWorkerPrisma } from "./shared"; +import { cleanupExpiredFitbitOAuthStates } from "./oauth-state-cleanup"; +import { getWorkerPrisma } from "@/lib/jobs/reminder/shared"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; /** diff --git a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts index 69ac042a1..32126000b 100644 --- a/src/lib/jobs/__tests__/integration-backfill-admission.test.ts +++ b/src/lib/jobs/__tests__/integration-backfill-admission.test.ts @@ -20,8 +20,8 @@ vi.mock("@/lib/jobs/whoop-backfill", async (importOriginal) => ({ enqueueBootTimeWhoopBackfill: async (stagger: number) => recordStagger("whoop-backfill", stagger), })); -vi.mock("@/lib/jobs/fitbit-backfill", async (importOriginal) => ({ - ...(await importOriginal()), +vi.mock("@/lib/fitbit/jobs/backfill", async (importOriginal) => ({ + ...(await importOriginal()), enqueueBootTimeFitbitBackfill: async (stagger: number) => recordStagger("fitbit-backfill", stagger), })); @@ -58,8 +58,8 @@ vi.mock("@/lib/jobs/strava-backfill", async (importOriginal) => ({ enqueueBootTimeStravaBackfill: async (stagger: number) => recordStagger("strava-backfill", stagger), })); -vi.mock("@/lib/jobs/fitbit-sleep-repair", async (importOriginal) => ({ - ...(await importOriginal()), +vi.mock("@/lib/fitbit/jobs/sleep-repair", async (importOriginal) => ({ + ...(await importOriginal()), enqueueBootTimeFitbitSleepRepair: async (stagger: number) => recordStagger("fitbit-sleep-repair", stagger), })); @@ -105,13 +105,13 @@ import { WHOOP_BACKFILL_QUEUE } from "../whoop-backfill-queue"; const providerSources = [ "whoop-backfill.ts", - "fitbit-backfill.ts", + "../fitbit/jobs/backfill.ts", "google-health-backfill.ts", "google-health-sleep-repair.ts", "sleep-timeline-backfill.ts", "lab-biomarker-backfill.ts", "strava-backfill.ts", - "fitbit-sleep-repair.ts", + "../fitbit/jobs/sleep-repair.ts", "oura-backfill.ts", ].map((file) => readFileSync(join(__dirname, "..", file), "utf8")); diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index e8b816cfb..127347574 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -1,7 +1,7 @@ /** * v1.26.0 — pg-boss queue + boot-time self-converging backfill for newly * connected Google Health accounts. Modelled on the Fitbit / WHOOP backfill - * (`src/lib/jobs/fitbit-backfill.ts`): a discovery query enqueues one job per + * (`src/lib/fitbit/jobs/backfill.ts`): a discovery query enqueues one job per * connection that has NOT yet been backfilled, and the pass is idempotent * across reboots (the predicate `backfill_completed_at IS NULL` drops a * connection once its backfill finishes). diff --git a/src/lib/jobs/integration-backfill-drain.ts b/src/lib/jobs/integration-backfill-drain.ts index dbce7a548..033002cb6 100644 --- a/src/lib/jobs/integration-backfill-drain.ts +++ b/src/lib/jobs/integration-backfill-drain.ts @@ -17,10 +17,10 @@ import type { Job } from "pg-boss"; import { runWhoopBackfillForUser } from "@/lib/jobs/whoop-backfill"; -import { runFitbitBackfillForUser } from "@/lib/jobs/fitbit-backfill"; +import { runFitbitBackfillForUser } from "@/lib/fitbit/jobs/backfill"; import { runGoogleHealthBackfillForUser } from "@/lib/jobs/google-health-backfill"; import { runGoogleHealthSleepRepairForUser } from "@/lib/jobs/google-health-sleep-repair"; -import { runFitbitSleepRepairForUser } from "@/lib/jobs/fitbit-sleep-repair"; +import { runFitbitSleepRepairForUser } from "@/lib/fitbit/jobs/sleep-repair"; import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill"; import { runOuraBackfillForUser } from "@/lib/jobs/oura-backfill"; import { runSleepTimelineBackfillForUser } from "@/lib/jobs/sleep-timeline-backfill"; diff --git a/src/lib/jobs/sleep-timeline-backfill.ts b/src/lib/jobs/sleep-timeline-backfill.ts index 2adbfa5d5..2bec8fae6 100644 --- a/src/lib/jobs/sleep-timeline-backfill.ts +++ b/src/lib/jobs/sleep-timeline-backfill.ts @@ -25,7 +25,7 @@ * A reboot mid-pass re-runs delete + re-sync from scratch — the re-sync upserts * are key-stable, so the result converges. * - * Modelled on `whoop-backfill.ts` / `fitbit-backfill.ts`. The queue name MUST + * Modelled on `whoop-backfill.ts` / `fitbit/jobs/backfill.ts`. The queue name MUST * be registered in `allQueues` in `src/lib/jobs/reminder-worker.ts` or pg-boss * never provisions it and the boot enqueue silently never drains (the v1.4.37 * dead-queue class).