chore(queue): schedule geo-backfill via pg-boss on the :40 hourly slot

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.
This commit is contained in:
Marc-André Bombeck committed 2026-05-17 14:42:09 +02:00
1 parent c824488895
commit d3c9eaaf48
3 files changed
+99

No files matched your search

@@ -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");
});
});
+12
View File
@@ -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;
+49
View File
@@ -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<GeoBackfillPayload>[]) {
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<PrDetectionPayload | { userId?: undefined }>[],
) {
@@ -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<GeoBackfillPayload>(
GEO_BACKFILL_QUEUE,
{ localConcurrency: 1 },
handleGeoBackfill,
);
await boss.work<PrDetectionPayload>(
PR_DETECTION_QUEUE,
{ localConcurrency: PR_DETECTION_CONCURRENCY },