From d3c9eaaf4863ce7a833e4aacc814cea6bf708bfd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 17 May 2026 14:42:09 +0200 Subject: [PATCH] chore(queue): schedule geo-backfill via pg-boss on the :40 hourly slot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `runGeoBackfill` helper has shipped since v1.4.27 but was never wired into the worker boot — it only ever fired through a test path, so the admin sign-in overview's "Standort" column kept rendering "—" on every historical row where the fire-and-forget IP resolver missed at write time. Marc's report against v1.4.36 made the problem visible: even after the v1.4.36 fallback fix, the backfill was missing entirely. Wire the helper into the existing pg-boss schedule list at :40 every hour. The queue name + cron expression live in the helper module so the worker boot pulls them by import; a co-located unit test pins both shapes (queue name + cron slot) plus a source-text probe that confirms the worker actually imports and registers the runner. The helper itself is unchanged — it stays idempotent, capped at 5 000 rows per pass, and scoped to the trailing 30-day window that the operational triage cares about. The :40 slot was chosen to dodge the existing :00 (Withings activity + measure crons), :15 (Withings sleep), :30 (moodlog sync), and :25 (medication compliance status) crowd at the top of the hour; the helper's per-call cap keeps the audit-log write budget under control even if a future ingest path enqueues a hot run. --- src/lib/jobs/__tests__/geo-backfill.test.ts | 38 ++++++++++++++++ src/lib/jobs/geo-backfill.ts | 12 +++++ src/lib/jobs/reminder-worker.ts | 49 +++++++++++++++++++++ 3 files changed, 99 insertions(+) diff --git a/src/lib/jobs/__tests__/geo-backfill.test.ts b/src/lib/jobs/__tests__/geo-backfill.test.ts index 9f835ce3e..f34493fff 100644 --- a/src/lib/jobs/__tests__/geo-backfill.test.ts +++ b/src/lib/jobs/__tests__/geo-backfill.test.ts @@ -8,6 +8,8 @@ vi.mock("@/lib/geo", () => ({ import { GEO_BACKFILL_BATCH_CAP, + GEO_BACKFILL_CRON, + GEO_BACKFILL_QUEUE, GEO_BACKFILL_WINDOW_DAYS, runGeoBackfill, } from "../geo-backfill"; @@ -217,3 +219,39 @@ describe("runGeoBackfill", () => { expect(prisma.auditLog.update).not.toHaveBeenCalled(); }); }); + +/** + * v1.4.37 — the helper is now scheduled via pg-boss at the hourly + * :40 slot. The schedule constants live in the helper module so the + * reminder-worker boot pulls them by import; this test pins both + * shapes so a regression that drops the cadence or drifts the queue + * name lands here, not in production. + */ +describe("geo-backfill scheduling contract (v1.4.37)", () => { + it("exports the pg-boss queue name", () => { + expect(GEO_BACKFILL_QUEUE).toBe("geo-backfill"); + }); + + it("schedules on the :40 slot every hour to avoid colliding with the existing crons", () => { + expect(GEO_BACKFILL_CRON).toBe("40 * * * *"); + }); + + it("reminder-worker imports both the runner and the schedule constants", async () => { + // Source-text probe avoids importing the worker module (which + // would drag pg-boss + a live Postgres connection). The contract + // is binary: the worker either registers the queue + cron or it + // does not; both presence checks pin the v1.4.37 scheduling + // shape end-to-end. + const fs = await import("node:fs/promises"); + const path = await import("node:path"); + const workerSrc = await fs.readFile( + path.resolve(__dirname, "..", "reminder-worker.ts"), + "utf-8", + ); + expect(workerSrc).toContain( + "import {\n runGeoBackfill,\n GEO_BACKFILL_QUEUE,\n GEO_BACKFILL_CRON,\n} from \"@/lib/jobs/geo-backfill\";", + ); + expect(workerSrc).toContain("GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON"); + expect(workerSrc).toContain("handleGeoBackfill"); + }); +}); diff --git a/src/lib/jobs/geo-backfill.ts b/src/lib/jobs/geo-backfill.ts index 071247c19..2ab37decc 100644 --- a/src/lib/jobs/geo-backfill.ts +++ b/src/lib/jobs/geo-backfill.ts @@ -38,6 +38,18 @@ import { lookupIpAsn, lookupIpLocation } from "@/lib/geo"; export const GEO_BACKFILL_BATCH_CAP = 5000; export const GEO_BACKFILL_WINDOW_DAYS = 30; +/** + * v1.4.37 — pg-boss queue name + cron expression for the recurring + * backfill. Offset to :40 every hour so the schedule does not collide + * with the existing :00 / :15 / :30 Withings + moodlog crons that + * already crowd the top of the hour. The helper is idempotent + the + * batch cap is per-call, so an hourly cadence is the right balance + * between freshness of the admin sign-in "Standort" column and the + * audit-log write budget. + */ +export const GEO_BACKFILL_QUEUE = "geo-backfill"; +export const GEO_BACKFILL_CRON = "40 * * * *"; + export interface GeoBackfillSummary { scanned: number; located: number; diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 9ba0661ad..eab758779 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -33,6 +33,11 @@ import { cleanupExpiredIdempotencyKeys } from "@/lib/jobs/idempotency-cleanup"; import { cleanupOldAuditLogs } from "@/lib/jobs/audit-log-cleanup"; import { runHostMetricTick } from "@/lib/jobs/host-metric-sampler"; import { aggregateRecommendationFeedback } from "@/lib/jobs/feedback-aggregator"; +import { + runGeoBackfill, + GEO_BACKFILL_QUEUE, + GEO_BACKFILL_CRON, +} from "@/lib/jobs/geo-backfill"; import { PR_DETECTION_QUEUE, PR_DETECTION_CONCURRENCY, @@ -158,6 +163,9 @@ const HOST_METRIC_CRON = "* * * * *"; // audit-log) so the previous-day's noise is gone before we aggregate. const FEEDBACK_AGGREGATOR_QUEUE = "feedback-aggregator"; const FEEDBACK_AGGREGATOR_CRON = "0 4 * * *"; +// v1.4.37 — hourly geo backfill. Queue name + cron expression live +// in `@/lib/jobs/geo-backfill` so a unit test can pin the scheduling +// contract without importing this worker boot file. // v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative // rows. Collapses each user × cumulative-type × calendar-day bucket // into one `stats:…` row so the list view stops painting hundreds of @@ -263,6 +271,10 @@ interface FeedbackAggregatorPayload { triggeredAt: string; } +interface GeoBackfillPayload { + triggeredAt: string; +} + // Re-export timezone utilities under local names for backward compatibility const getUserTodayBounds = getUserTodayBoundsUtil; const getDayOfWeekInTz = getDayOfWeekInTzUtil; @@ -1170,6 +1182,32 @@ async function handleFeedbackAggregator( }); } +/** + * v1.4.37 — geo-backfill worker. Walks `audit_logs` rows that landed + * with a null `location` (offline MMDB missing at write time, online + * provider unreachable) and re-resolves them through the now-bundled + * resolver chain. The helper is idempotent and capped per pass so + * the hourly cadence cannot starve a live login spike. + */ +async function handleGeoBackfill(jobs: Job[]) { + void jobs; + await withBackgroundEvent("job.geo_backfill", async (evt) => { + const p = getWorkerPrisma(); + try { + const summary = await runGeoBackfill(p); + evt.addMeta("geo_backfill_scanned", summary.scanned); + evt.addMeta("geo_backfill_located", summary.located); + evt.addMeta("geo_backfill_carrier_resolved", summary.carrierResolved); + evt.addMeta("geo_backfill_still_unresolved", summary.stillUnresolved); + } catch (err) { + // The admin sign-in overview tolerates a stale Standort cell — + // log and move on so a one-off resolver hiccup does not poison + // the queue and block the next pass. + evt.addWarning(`geo-backfill failed: ${err}`); + } + }); +} + async function handlePrDetection( jobs: Job[], ) { @@ -1481,6 +1519,7 @@ export async function startReminderWorker() { OFFHOST_BACKUP_QUEUE, HOST_METRIC_QUEUE, FEEDBACK_AGGREGATOR_QUEUE, + GEO_BACKFILL_QUEUE, PR_DETECTION_QUEUE, MEDICATION_INVENTORY_EXPIRE_QUEUE, APPLE_HEALTH_IMPORT_QUEUE, @@ -1526,6 +1565,11 @@ export async function startReminderWorker() { [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON], [HOST_METRIC_QUEUE, HOST_METRIC_CRON], [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON], + // v1.4.37 — hourly geo backfill. The helper is idempotent + capped + // at 5 000 rows per pass; running it at :40 every hour catches the + // long tail of audit rows that landed with the offline MMDB + // missing or the online provider unreachable. + [GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON], // Fallback rescan every 30 minutes — protects against ingest paths // that ship measurements without enqueueing a per-user job. The // cron payload deliberately omits a `userId` so the handler iterates @@ -1643,6 +1687,11 @@ export async function startReminderWorker() { { localConcurrency: 1 }, handleFeedbackAggregator, ); + await boss.work( + GEO_BACKFILL_QUEUE, + { localConcurrency: 1 }, + handleGeoBackfill, + ); await boss.work( PR_DETECTION_QUEUE, { localConcurrency: PR_DETECTION_CONCURRENCY },