mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 21:10:59 +00:00
feat(vorsorge): resolve reminders on ingest, not only on the cron
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.
This commit is contained in:
1 parent
f92247760c
commit
7b57c1d3d8
14 files changed
+389
-5
No files matched your search
@@ -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);
|
||||
}
|
||||
@@ -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<Response> {
|
||||
// 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", {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -672,6 +672,10 @@ export async function runFitbitPollCohort(
|
||||
concurrency?: number;
|
||||
sync?: (userId: string) => Promise<number>;
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -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/,
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> }> = [];
|
||||
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<DispatchFn>(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<DispatchFn>(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: [
|
||||
|
||||
@@ -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<Record<string, unknown>>;
|
||||
measurement?: { measuredAt: Date } | null;
|
||||
lab?: { takenAt: Date } | null;
|
||||
}) {
|
||||
const updates: Array<{ id: string; data: Record<string, unknown> }> = [];
|
||||
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<string, unknown> }) => {
|
||||
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<string, unknown> = {}) {
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<void> {
|
||||
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,
|
||||
});
|
||||
}
|
||||
@@ -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<ReminderSatisfyPayload>(
|
||||
REMINDER_SATISFY_QUEUE,
|
||||
{ localConcurrency: REMINDER_SATISFY_CONCURRENCY },
|
||||
handleReminderSatisfy,
|
||||
);
|
||||
await boss.work<MedicationInventoryExpirePayload>(
|
||||
MEDICATION_INVENTORY_EXPIRE_QUEUE,
|
||||
{ localConcurrency: 1 },
|
||||
|
||||
@@ -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<FitbitSyncPayload>[]) {
|
||||
{
|
||||
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(() => {});
|
||||
}
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
@@ -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<ReminderSatisfyPayload>[],
|
||||
) {
|
||||
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;
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -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}`);
|
||||
}
|
||||
|
||||
@@ -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}`);
|
||||
}
|
||||
|
||||
@@ -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}`,
|
||||
|
||||
Reference in new issue
Block a user