mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-11 13:52:20 +00:00
refactor(fitbit): keep the Fitbit jobs inside the Fitbit folder
The backfill, the sleep repair, the OAuth-state cleanup and the sync handlers move from lib/jobs into lib/fitbit/jobs, beside the queue facts and definitions the registry already spreads. Their tests move with them. Removing the integration is now the folder, its routes, the registry spread and the drain's two cases. The updatedAt backfill guard also walks provider jobs folders, so the moved backfill stays covered.
This commit is contained in:
1 parent
7e54b6be81
commit
8b3c095df5
16 files changed
+41
-37
No files matched your search
@@ -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/<provider>/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<string, string> = {
|
||||
"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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
+1
-1
@@ -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(() => {
|
||||
+2
-2
@@ -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",
|
||||
);
|
||||
|
||||
+1
-1
@@ -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(() => {
|
||||
+1
-1
@@ -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", () => {
|
||||
@@ -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;
|
||||
@@ -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"),
|
||||
),
|
||||
},
|
||||
|
||||
File renamed without changes.
@@ -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;
|
||||
@@ -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";
|
||||
|
||||
/**
|
||||
@@ -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<typeof import("@/lib/jobs/fitbit-backfill")>()),
|
||||
vi.mock("@/lib/fitbit/jobs/backfill", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("@/lib/fitbit/jobs/backfill")>()),
|
||||
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<typeof import("@/lib/jobs/fitbit-sleep-repair")>()),
|
||||
vi.mock("@/lib/fitbit/jobs/sleep-repair", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("@/lib/fitbit/jobs/sleep-repair")>()),
|
||||
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"));
|
||||
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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).
|
||||
|
||||
Reference in new issue
Block a user