From 7b57c1d3d84f3223d427a345374fddcdda3468e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 16 Jun 2026 20:58:56 +0200 Subject: [PATCH] feat(vorsorge): resolve reminders on ingest, not only on the cron MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a reminder-satisfy pg-boss queue (mirrors pr-detection) that every measurement and lab ingest path enqueues — manual create, batch, and the Withings / Polar / Oura / Nightscout / Fitbit / WHOOP syncs. A reminder now auto-resolves the moment a matching reading lands instead of waiting up to 15 minutes. The 15-minute cron stays as the idempotent safety net; the forward-only guard lets the eventful hook and the cron converge without double-stamping. The worker honours the module toggle (a reminder for a disabled module produces no engine activity) and runs the same satisfaction primitives the cron uses. --- src/app/api/labs/route.ts | 6 + src/app/api/measurements/batch/route.ts | 5 + src/app/api/measurements/route.ts | 9 ++ src/lib/fitbit/sync.ts | 5 + .../measurement-reminder-queue.test.ts | 22 +++ .../__tests__/measurement-reminder.test.ts | 35 +++- .../jobs/__tests__/reminder-satisfy.test.ts | 153 ++++++++++++++++++ src/lib/jobs/reminder-satisfy.ts | 54 +++++++ src/lib/jobs/reminder-worker.ts | 19 +++ src/lib/jobs/reminder/fitbit-sync.ts | 8 + src/lib/jobs/reminder/mood-cycle-checks.ts | 48 +++++- src/lib/jobs/reminder/poll-cohort.ts | 9 +- src/lib/jobs/reminder/whoop-sync.ts | 9 +- src/lib/jobs/reminder/withings-sync.ts | 12 ++ 14 files changed, 389 insertions(+), 5 deletions(-) create mode 100644 src/lib/jobs/__tests__/reminder-satisfy.test.ts create mode 100644 src/lib/jobs/reminder-satisfy.ts diff --git a/src/app/api/labs/route.ts b/src/app/api/labs/route.ts index a80ff3f41..2a8ec3902 100644 --- a/src/app/api/labs/route.ts +++ b/src/app/api/labs/route.ts @@ -11,6 +11,7 @@ import { import { auditLog } from "@/lib/auth/audit"; import { prisma } from "@/lib/db"; import { withIdempotency } from "@/lib/idempotency"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { encryptNoteToBytes } from "@/lib/labs/store"; import { annotate } from "@/lib/logging/context"; import { @@ -184,5 +185,10 @@ async function postLabResult(request: NextRequest) { meta: { labResultId: created.id }, }); + // v1.18.1 (D2) — eventful Lab↔Vorsorge satisfaction. A lab panel just + // landed; resolve the user's free-text "annual blood panel" reminders + // now rather than waiting on the 15-min cron. Fire-and-forget. + void enqueueReminderSatisfy(user.id).catch(() => {}); + return apiSuccess(serialiseLabResult(created), 201); } diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 16aea90fe..849d78c5a 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -52,6 +52,7 @@ import { validateMeasurementRange } from "@/lib/validations/measurement"; import { deviceTypeEnum } from "@/lib/validations/source-priority"; import { checkRateLimit } from "@/lib/rate-limit"; import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { @@ -599,6 +600,10 @@ async function postBatch(request: NextRequest): Promise { // personal record. if (insertedCount > 0 || updatedCount > 0 || duplicateCount > 0) { const silent = entries.length > PR_DETECTION_SILENT_THRESHOLD; + // v1.18.1 — eventful Vorsorge satisfaction. A matching reading just + // landed; resolve the user's reminders now rather than waiting on the + // 15-min cron. Fire-and-forget — the cron reconciles on enqueue miss. + void enqueueReminderSatisfy(user.id).catch(() => {}); try { await enqueuePrDetection(user.id, { silent }); await auditLog("personal_records.detection_enqueued", { diff --git a/src/app/api/measurements/route.ts b/src/app/api/measurements/route.ts index 4490aaa3d..57f78876c 100644 --- a/src/app/api/measurements/route.ts +++ b/src/app/api/measurements/route.ts @@ -29,6 +29,7 @@ import { import { withIdempotency } from "@/lib/idempotency"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { recomputeBucketsForMeasurement, collapseToTypeDayKeys, @@ -924,6 +925,10 @@ async function postMeasurement(request: NextRequest) { // SWR readers don't serve the pre-write body back to the user. invalidateUserMeasurements(user.id, { evict: true }); + // v1.18.1 — eventful Vorsorge satisfaction. Resolve the user's + // reminders against the just-landed readings now. Fire-and-forget. + void enqueueReminderSatisfy(user.id).catch(() => {}); + // v1.5.0 — refresh the persistent rollup table for every distinct // (type, day) the batch touched so the next analytics / coach read // hits the cache rather than falling through to live aggregation. @@ -1037,6 +1042,10 @@ async function postMeasurement(request: NextRequest) { // pre-write body back to the user. invalidateUserMeasurements(user.id, { evict: true }); + // v1.18.1 — eventful Vorsorge satisfaction. Resolve the user's reminders + // against the just-landed reading now. Fire-and-forget. + void enqueueReminderSatisfy(user.id).catch(() => {}); + // v1.5.0 — refresh the persistent rollup row for the affected // (type, day) tuple. Runs inline so the next read of the // comprehensive / analytics surface is correct without waiting on diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index c50525594..2c4c02059 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -672,6 +672,10 @@ export async function runFitbitPollCohort( concurrency?: number; sync?: (userId: string) => Promise; onUserError?: (userId: string, err: unknown) => void; + /** v1.18.1 — invoked after a user's sync lands `imported` measurements, + * so the worker can fire the eventful Vorsorge satisfaction enqueue + * without the fitbit lib reaching into the job layer. */ + onUserSynced?: (userId: string, imported: number) => void; } = {}, ): Promise<{ usersSynced: number; measurementsImported: number }> { const sync = opts.sync ?? ((userId: string) => syncUserFitbit(userId)); @@ -689,6 +693,7 @@ export async function runFitbitPollCohort( const n = await sync(userId); measurementsImported = measurementsImported + n; usersSynced = usersSynced + 1; + opts.onUserSynced?.(userId, n); } catch (err) { opts.onUserError?.(userId, err); } diff --git a/src/lib/jobs/__tests__/measurement-reminder-queue.test.ts b/src/lib/jobs/__tests__/measurement-reminder-queue.test.ts index e159e626f..8eb1ac19b 100644 --- a/src/lib/jobs/__tests__/measurement-reminder-queue.test.ts +++ b/src/lib/jobs/__tests__/measurement-reminder-queue.test.ts @@ -58,3 +58,25 @@ describe("reminder-worker — measurement-reminder wiring", () => { ); }); }); + +describe("reminder-worker — eventful reminder-satisfy wiring (v1.18.1)", () => { + it("imports the satisfy queue constants", () => { + expect(source).toMatch(/from\s*["']@\/lib\/jobs\/reminder-satisfy["']/); + expect(source).toMatch(/\bREMINDER_SATISFY_QUEUE\b/); + }); + + it("registers the satisfy queue in the allQueues loop", () => { + const allQueuesMatch = source.match(/const allQueues\s*=\s*\[([\s\S]*?)\];/); + expect(allQueuesMatch).not.toBeNull(); + expect(allQueuesMatch![1]).toMatch(/\bREMINDER_SATISFY_QUEUE\b/); + }); + + it("binds a boss.work handler for the satisfy queue", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,200}REMINDER_SATISFY_QUEUE[\s\S]{0,200}handleReminderSatisfy/, + ); + expect(source).toMatch( + /handleReminderSatisfy[\s\S]{0,600}runReminderSatisfyForUser/, + ); + }); +}); diff --git a/src/lib/jobs/__tests__/measurement-reminder.test.ts b/src/lib/jobs/__tests__/measurement-reminder.test.ts index 22f3ad6fe..49dff77d3 100644 --- a/src/lib/jobs/__tests__/measurement-reminder.test.ts +++ b/src/lib/jobs/__tests__/measurement-reminder.test.ts @@ -126,6 +126,9 @@ interface FakeReminder { function makePrisma(opts: { reminders: FakeReminder[]; measurementMatch?: { measuredAt: Date } | null; + /** v1.18.1 (D2) — a free-text reminder now auto-resolves from a matching + * LabResult. Default `null` (no lab landed). */ + labMatch?: { takenAt: Date } | null; }) { const updates: Array<{ id: string; data: Record }> = []; const prisma = { @@ -147,6 +150,9 @@ function makePrisma(opts: { measurement: { findFirst: vi.fn(async () => opts.measurementMatch ?? null), }, + labResult: { + findFirst: vi.fn(async () => opts.labMatch ?? null), + }, }; return { prisma, updates }; } @@ -220,11 +226,13 @@ describe("runMeasurementReminderTick", () => { expect(advanced.getTime()).toBeGreaterThan(NINE_LOCAL.getTime()); }); - it("free-text reminder never auto-resolves (no measurement query match path)", async () => { + it("free-text reminder never queries Measurement (matches on LabResult instead)", async () => { const { prisma } = makePrisma({ reminders: [reminder({ measurementType: null })], - // Even if a reading existed, a free-text reminder must not query it. + // Even if a reading existed, a free-text reminder must not query it — + // it resolves from a LabResult (D2), and none landed here. measurementMatch: { measuredAt: new Date("2026-06-14T18:00:00Z") }, + labMatch: null, }); const dispatch = vi.fn(async () => OK); @@ -239,6 +247,29 @@ describe("runMeasurementReminderTick", () => { expect(prisma.measurement.findFirst).not.toHaveBeenCalled(); }); + it("v1.18.1 (D2) — free-text reminder auto-resolves from a matching LabResult", async () => { + const takenAt = new Date("2026-06-14T18:00:00Z"); + const { prisma, updates } = makePrisma({ + reminders: [reminder({ measurementType: null })], + labMatch: { takenAt }, + }); + const dispatch = vi.fn(async () => OK); + + const summary = await runMeasurementReminderTick( + prisma as never, + NINE_LOCAL, + { dispatch }, + ); + + expect(summary.autoResolved).toBe(1); + expect(summary.dispatched).toBe(0); + expect(dispatch).not.toHaveBeenCalled(); + expect(updates).toHaveLength(1); + expect(updates[0].data.lastSatisfiedAt).toEqual(takenAt); + // A free-text reminder resolves from labs, never from measurements. + expect(prisma.measurement.findFirst).not.toHaveBeenCalled(); + }); + it("suppresses the push under clientManaged but still advances nextDueAt", async () => { const { prisma, updates } = makePrisma({ reminders: [ diff --git a/src/lib/jobs/__tests__/reminder-satisfy.test.ts b/src/lib/jobs/__tests__/reminder-satisfy.test.ts new file mode 100644 index 000000000..5320c618b --- /dev/null +++ b/src/lib/jobs/__tests__/reminder-satisfy.test.ts @@ -0,0 +1,153 @@ +/** + * v1.18.1 — eventful Vorsorge satisfaction. + * + * 1. `enqueueReminderSatisfy` — a thin wrapper around the global boss, + * a silent no-op when no worker is attached (mirrors enqueuePrDetection). + * 2. `runReminderSatisfyForUser` — resolves a user's live reminders + * against their just-landed measurement / lab via the shared + * primitives, honouring the module toggle and the forward-only guard. + */ +import { describe, expect, it, vi, beforeEach } from "vitest"; + +const sendMock = vi.fn(); +let bossInstance: { send: typeof sendMock } | null = { send: sendMock }; + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => bossInstance, +})); + +import { enqueueReminderSatisfy, REMINDER_SATISFY_QUEUE } from "../reminder-satisfy"; +import { runReminderSatisfyForUser } from "../measurement-reminder"; + +describe("enqueueReminderSatisfy", () => { + beforeEach(() => { + sendMock.mockReset(); + bossInstance = { send: sendMock }; + }); + + it("sends a job carrying the userId to the satisfy queue", async () => { + await enqueueReminderSatisfy("u1"); + expect(sendMock).toHaveBeenCalledTimes(1); + expect(sendMock.mock.calls[0]![0]).toBe(REMINDER_SATISFY_QUEUE); + expect(sendMock.mock.calls[0]![1]).toMatchObject({ userId: "u1" }); + }); + + it("is a silent no-op when no boss is attached (tests / web process)", async () => { + bossInstance = null; + await expect(enqueueReminderSatisfy("u1")).resolves.toBeUndefined(); + expect(sendMock).not.toHaveBeenCalled(); + }); +}); + +const TZ = "Europe/Berlin"; + +function makePrisma(opts: { + reminders: Array>; + measurement?: { measuredAt: Date } | null; + lab?: { takenAt: Date } | null; +}) { + const updates: Array<{ id: string; data: Record }> = []; + const prisma = { + user: { findUnique: vi.fn(async () => ({ timezone: TZ })) }, + measurementReminder: { + findMany: vi.fn(async () => opts.reminders), + update: vi.fn( + async ({ where, data }: { where: { id: string }; data: Record }) => { + updates.push({ id: where.id, data }); + return { id: where.id, ...data }; + }, + ), + }, + measurement: { findFirst: vi.fn(async () => opts.measurement ?? null) }, + labResult: { findFirst: vi.fn(async () => opts.lab ?? null) }, + }; + return { prisma, updates }; +} + +function reminderRow(overrides: Record = {}) { + return { + id: "r1", + measurementType: "WEIGHT", + intervalDays: 7, + rrule: null, + anchorDate: null, + notifyHour: 9, + lastSatisfiedAt: null, + createdAt: new Date("2026-06-01T00:00:00Z"), + ...overrides, + }; +} + +describe("runReminderSatisfyForUser", () => { + it("satisfies a typed reminder eventfully when a matching reading landed", async () => { + const measuredAt = new Date("2026-06-14T18:00:00Z"); + const { prisma, updates } = makePrisma({ + reminders: [reminderRow()], + measurement: { measuredAt }, + }); + + const summary = await runReminderSatisfyForUser( + prisma as never, + "u1", + new Date("2026-06-14T20:00:00Z"), + { isModuleEnabled: vi.fn(async () => true) }, + ); + + expect(summary.satisfied).toBe(1); + expect(updates).toHaveLength(1); + expect(updates[0].data.lastSatisfiedAt).toEqual(measuredAt); + }); + + it("satisfies a free-text reminder from a LabResult (D2)", async () => { + const takenAt = new Date("2026-06-15T09:00:00Z"); + const { prisma, updates } = makePrisma({ + reminders: [reminderRow({ measurementType: null, intervalDays: 365 })], + lab: { takenAt }, + }); + + const summary = await runReminderSatisfyForUser( + prisma as never, + "u1", + new Date("2026-06-15T12:00:00Z"), + ); + + expect(summary.satisfied).toBe(1); + expect(updates[0].data.lastSatisfiedAt).toEqual(takenAt); + }); + + it("produces no engine activity for a disabled module", async () => { + const { prisma, updates } = makePrisma({ + reminders: [reminderRow({ measurementType: "BLOOD_GLUCOSE" })], + measurement: { measuredAt: new Date("2026-06-14T18:00:00Z") }, + }); + const isModuleEnabled = vi.fn(async () => false); + + const summary = await runReminderSatisfyForUser( + prisma as never, + "u1", + new Date(), + { isModuleEnabled }, + ); + + expect(summary.skippedModuleDisabled).toBe(1); + expect(summary.satisfied).toBe(0); + expect(isModuleEnabled).toHaveBeenCalledWith("u1", "glucose"); + expect(updates).toHaveLength(0); + // The matcher never runs for a gated reminder. + expect(prisma.measurement.findFirst).not.toHaveBeenCalled(); + }); + + it("is forward-only: a stale event is a no-op (cron + hook converge)", async () => { + const { prisma, updates } = makePrisma({ + reminders: [ + reminderRow({ lastSatisfiedAt: new Date("2026-06-20T08:00:00Z") }), + ], + measurement: { measuredAt: new Date("2026-06-10T08:00:00Z") }, + }); + + const summary = await runReminderSatisfyForUser(prisma as never, "u1", new Date()); + + expect(summary.satisfied).toBe(0); + expect(updates).toHaveLength(0); + }); +}); diff --git a/src/lib/jobs/reminder-satisfy.ts b/src/lib/jobs/reminder-satisfy.ts new file mode 100644 index 000000000..b06e511cb --- /dev/null +++ b/src/lib/jobs/reminder-satisfy.ts @@ -0,0 +1,54 @@ +/** + * v1.18.1 — enqueue helper for the eventful Vorsorge-reminder satisfaction + * queue. Mirrors `pr-detection.ts`: the Next.js ingest routes import + * `enqueueReminderSatisfy` and never touch pg-boss directly, the worker + * binds the handler in `reminder-worker.ts`, and the 15-min + * measurement-reminder cron is the idempotent safety-net for any ingest + * path that forgets to enqueue (or any job that fails to land). + * + * Why a queue and not an inline call on the ingest path: the batch route's + * own comment warns against hooking the hot iOS batch-ingest path. The + * matcher re-queries the user's measurements/labs, so keep it off the + * request and out of the sync worker's tight loop — enqueue + * fire-and-forget, resolve in the worker. + */ +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; + +export const REMINDER_SATISFY_QUEUE = "reminder-satisfy"; + +/** Concurrency budget for the worker process. The job is read-heavy on the + * user's own reminders + a single indexed measurement/lab lookup per + * typed reminder, so a small fleet drains an ingest spike without + * crowding the reminder check or the daily insights workload on the same + * node. */ +export const REMINDER_SATISFY_CONCURRENCY = 5; + +export interface ReminderSatisfyPayload { + userId: string; + /** ISO timestamp — handy for debugging, never used for logic. */ + triggeredAt: string; +} + +/** + * Submit a reminder-satisfaction job for one user after a measurement / + * lab write. Best-effort: when no boss instance is attached (the route + * runs in a context without the worker — typically tests, scripts), the + * call is a silent no-op and the 15-min cron reconciles the user within + * one tick. + */ +export async function enqueueReminderSatisfy(userId: string): Promise { + const boss = getGlobalBoss(); + if (!boss) return; + const payload: ReminderSatisfyPayload = { + userId, + triggeredAt: new Date().toISOString(), + }; + await boss.send(REMINDER_SATISFY_QUEUE, payload, { + // Gentle retry — the resolver is read-heavy on the user's own data; a + // stampede of failed jobs against a briefly-flaky DB would compound the + // problem. The cron is the ultimate backstop anyway. + retryLimit: 3, + retryDelay: 30, + retryBackoff: true, + }); +} diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index db6b4d5b2..84fc7ce5e 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -42,6 +42,11 @@ import { PR_DETECTION_FALLBACK_CRON, type PrDetectionPayload, } from "@/lib/jobs/pr-detection"; +import { + REMINDER_SATISFY_QUEUE, + REMINDER_SATISFY_CONCURRENCY, + type ReminderSatisfyPayload, +} from "@/lib/jobs/reminder-satisfy"; import { MEDICATION_INVENTORY_EXPIRE_QUEUE, MEDICATION_INVENTORY_EXPIRE_CRON, @@ -250,6 +255,7 @@ import { handleMoodReminderCheck, handleCycleReminderCheck, handleMeasurementReminderCheck, + handleReminderSatisfy, } from "./reminder/mood-cycle-checks"; import { HostMetricSamplePayload, @@ -677,6 +683,11 @@ export async function startReminderWorker() { // schedule below silently no-ops (the v1.4.37 dead-queue class). TLS_PIN_MONITOR_QUEUE, PR_DETECTION_QUEUE, + // v1.18.1 — eventful Vorsorge satisfaction. No cron of its own (the + // 15-min measurement-reminder check is the safety-net); enqueued by the + // ingest paths. Must still be registered here or the worker binding + // below never provisions the queue (v1.4.37 dead-queue lesson). + REMINDER_SATISFY_QUEUE, MEDICATION_INVENTORY_EXPIRE_QUEUE, // v1.4.46 — hourly auto-skip pass for stale unmarked intakes. // Same pg-boss v12 createQueue contract as the other crons; without @@ -1281,6 +1292,14 @@ export async function startReminderWorker() { { localConcurrency: PR_DETECTION_CONCURRENCY }, handlePrDetection, ); + // v1.18.1 — eventful Vorsorge satisfaction. Resolves a user's reminders + // against their just-landed measurement / lab. Read-heavy on the user's + // own data; the same small concurrency budget as PR detection. + await boss.work( + REMINDER_SATISFY_QUEUE, + { localConcurrency: REMINDER_SATISFY_CONCURRENCY }, + handleReminderSatisfy, + ); await boss.work( MEDICATION_INVENTORY_EXPIRE_QUEUE, { localConcurrency: 1 }, diff --git a/src/lib/jobs/reminder/fitbit-sync.ts b/src/lib/jobs/reminder/fitbit-sync.ts index 53e6ae00d..b591bfcd9 100644 --- a/src/lib/jobs/reminder/fitbit-sync.ts +++ b/src/lib/jobs/reminder/fitbit-sync.ts @@ -8,6 +8,7 @@ import { type Job } from "pg-boss"; import { recordError } from "@/lib/jobs/worker-status"; 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"; @@ -45,6 +46,13 @@ export async function handleFitbitSync(jobs: Job[]) { { onUserError: (userId, err) => evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`), + // v1.18.1 — a fresh reading landed; resolve the user's Vorsorge + // reminders eventfully. Fire-and-forget; the cron is the net. + onUserSynced: (userId, imported) => { + if (imported > 0) { + void enqueueReminderSatisfy(userId).catch(() => {}); + } + }, }, ); diff --git a/src/lib/jobs/reminder/mood-cycle-checks.ts b/src/lib/jobs/reminder/mood-cycle-checks.ts index 655a5b2b3..b9cd2dbcd 100644 --- a/src/lib/jobs/reminder/mood-cycle-checks.ts +++ b/src/lib/jobs/reminder/mood-cycle-checks.ts @@ -9,7 +9,11 @@ import { recordError } from "@/lib/jobs/worker-status"; import { withBackgroundEvent } from "@/lib/logging/background"; import { runMoodReminderTick } from "@/lib/jobs/mood-reminder"; import { runCycleReminderTick } from "@/lib/jobs/cycle-reminder"; -import { runMeasurementReminderTick } from "@/lib/jobs/measurement-reminder"; +import { + runMeasurementReminderTick, + runReminderSatisfyForUser, +} from "@/lib/jobs/measurement-reminder"; +import type { ReminderSatisfyPayload } from "@/lib/jobs/reminder-satisfy"; import { getWorkerPrisma } from "./shared"; export interface MoodReminderPayload { @@ -120,3 +124,45 @@ export async function handleMeasurementReminderCheck( } }); } + +/** + * v1.18.1 — eventful Vorsorge satisfaction. Enqueued by every measurement + * / lab ingest path (manual create, batch, device syncs) so a reminder + * auto-resolves the moment a matching reading lands rather than waiting up + * to 15 minutes for the cron. Batches one job per user; the resolver loads + * that user's live reminders and runs the shared satisfy primitive. + */ +export async function handleReminderSatisfy( + jobs: Job[], +) { + await withBackgroundEvent("job.reminder_satisfy", async (evt) => { + const prisma = getWorkerPrisma(); + const now = new Date(); + // Distinct users in this batch — a multi-source ingest spike can enqueue + // several jobs for one user; resolve each user once. + const userIds = Array.from( + new Set(jobs.map((j) => j.data?.userId).filter((id): id is string => !!id)), + ); + let satisfied = 0; + let candidates = 0; + try { + for (const userId of userIds) { + const summary = await runReminderSatisfyForUser(prisma, userId, now); + satisfied += summary.satisfied; + candidates += summary.candidatesScanned; + } + evt.setBackground({ + task_name: "job.reminder_satisfy", + result: { + users: userIds.length, + candidates_scanned: candidates, + satisfied, + }, + }); + } catch (err) { + evt.setError(err); + recordError(); + throw err; + } + }); +} diff --git a/src/lib/jobs/reminder/poll-cohort.ts b/src/lib/jobs/reminder/poll-cohort.ts index e07bdcc18..a0e086aca 100644 --- a/src/lib/jobs/reminder/poll-cohort.ts +++ b/src/lib/jobs/reminder/poll-cohort.ts @@ -21,6 +21,7 @@ import { type Job } from "pg-boss"; import { recordError } from "@/lib/jobs/worker-status"; import { withBackgroundEvent } from "@/lib/logging/background"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { getWorkerPrisma } from "./shared"; /** Every poll-cohort handler accepts an optional single-user payload. */ @@ -69,8 +70,14 @@ export function makePollCohortHandler({ let measurementsImported = 0; for (const { userId } of targets) { try { - measurementsImported += await syncUser(userId); + const imported = await syncUser(userId); + measurementsImported += imported; usersSynced++; + // v1.18.1 — a fresh reading landed; resolve this user's Vorsorge + // reminders eventfully. Fire-and-forget; the cron is the net. + if (imported > 0) { + void enqueueReminderSatisfy(userId).catch(() => {}); + } } catch (err) { evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } diff --git a/src/lib/jobs/reminder/whoop-sync.ts b/src/lib/jobs/reminder/whoop-sync.ts index aab16721d..92a1e2ec3 100644 --- a/src/lib/jobs/reminder/whoop-sync.ts +++ b/src/lib/jobs/reminder/whoop-sync.ts @@ -20,6 +20,7 @@ import { syncUserWorkout, syncWhoopWorkoutById, } from "@/lib/whoop/sync-workout"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { getWorkerPrisma } from "./shared"; /** @@ -91,11 +92,17 @@ export async function runWhoopResourceSync( let measurementsImported = 0; for (const { userId, resourceId } of targets) { try { - measurementsImported += + const imported = resourceId && byIdFn ? await byIdFn(userId, resourceId) : await syncFn(userId); + measurementsImported += imported; usersSynced++; + // v1.18.1 — a fresh reading landed; resolve this user's Vorsorge + // reminders eventfully. Fire-and-forget; the cron is the net. + if (imported > 0) { + void enqueueReminderSatisfy(userId).catch(() => {}); + } } catch (err) { evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } diff --git a/src/lib/jobs/reminder/withings-sync.ts b/src/lib/jobs/reminder/withings-sync.ts index 3360813e9..0ebf2d182 100644 --- a/src/lib/jobs/reminder/withings-sync.ts +++ b/src/lib/jobs/reminder/withings-sync.ts @@ -10,6 +10,7 @@ import { withBackgroundEvent } from "@/lib/logging/background"; import { syncUserMeasurements } from "@/lib/withings/sync"; import { syncUserActivity } from "@/lib/withings/sync-activity"; import { syncUserSleep } from "@/lib/withings/sync-sleep"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { getWorkerPrisma } from "./shared"; export interface WithingsSyncPayload { @@ -64,6 +65,11 @@ export async function handleWithingsFallbackSync( const imported = await syncUserMeasurements(connection.userId); usersSynced++; measurementsImported += imported; + // v1.18.1 — a fresh reading landed; resolve this user's Vorsorge + // reminders eventfully. Fire-and-forget; the cron is the net. + if (imported > 0) { + void enqueueReminderSatisfy(connection.userId).catch(() => {}); + } } catch (err) { evt.addWarning( `Fallback sync failed for user ${connection.userId}: ${err}`, @@ -131,6 +137,9 @@ export async function handleWithingsActivitySync( const imported = await syncUserActivity(userId); usersSynced++; measurementsImported += imported; + if (imported > 0) { + void enqueueReminderSatisfy(userId).catch(() => {}); + } } catch (err) { evt.addWarning( `Withings activity sync failed for user ${userId}: ${err}`, @@ -186,6 +195,9 @@ export async function handleWithingsSleepSync( const imported = await syncUserSleep(userId); usersSynced++; measurementsImported += imported; + if (imported > 0) { + void enqueueReminderSatisfy(userId).catch(() => {}); + } } catch (err) { evt.addWarning( `Withings sleep sync failed for user ${userId}: ${err}`,