From 227eb69b63705d0de243594161a0d2f46431932b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 18 May 2026 12:30:25 +0200 Subject: [PATCH] perf(medications-intake): consume compliance rollup tier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Swaps `/api/medications/intake?scope=compliance` onto the rollup reader behind a coverage probe — a user with intake events but no rollup rows still gets a correct response from the legacy live aggregator and triggers the boot backfill in the background so the next request lands on the rollup path. Every write site that mutates `MedicationIntakeEvent` (POST + PUT + DELETE + per-medication import, bulk batch, telegram callbacks, external API ingest, admin restore) now fires `recomputeMedicationComplianceForEvent` for the affected user-day. Bulk + import collapse the per-row hooks to one recompute per distinct `(medicationId, dayKey)` so a 500-row ingest still costs a bounded fold. Purge wipes the rollup partition alongside the intake-event deleteMany. Eliminates the 3.2 s cold mount the v1.4.38 perf analysis attributed to the unbounded event findMany. --- .../api/admin/backups/[id]/restore/route.ts | 4 ++ src/app/api/ingest/medication/route.ts | 15 +++++ .../[id]/intake/[eventId]/route.ts | 33 ++++++++++ .../medications/[id]/intake/import/route.ts | 31 +++++++++ .../medications/[id]/intake/purge/route.ts | 7 ++ src/app/api/medications/[id]/intake/route.ts | 11 ++++ src/app/api/medications/intake/bulk/route.ts | 40 +++++++++++ src/app/api/medications/intake/route.ts | 66 ++++++++++++++++++- src/app/api/telegram/webhook/route.ts | 43 +++++++++++- 9 files changed, 246 insertions(+), 4 deletions(-) diff --git a/src/app/api/admin/backups/[id]/restore/route.ts b/src/app/api/admin/backups/[id]/restore/route.ts index 47b100828..e3c4e69f4 100644 --- a/src/app/api/admin/backups/[id]/restore/route.ts +++ b/src/app/api/admin/backups/[id]/restore/route.ts @@ -44,6 +44,10 @@ import { type BackupSummary, } from "@/lib/validations/backup"; import { recomputeUserMoodRollups } from "@/lib/mood/rollups"; +import { + recomputeUserMedicationCompliance, + MEDICATION_COMPLIANCE_BACKFILL_DAYS, +} from "@/lib/medications/compliance-rollups"; export const dynamic = "force-dynamic"; diff --git a/src/app/api/ingest/medication/route.ts b/src/app/api/ingest/medication/route.ts index b3c61e3c7..614502020 100644 --- a/src/app/api/ingest/medication/route.ts +++ b/src/app/api/ingest/medication/route.ts @@ -12,6 +12,7 @@ import { hashToken } from "@/lib/auth/hmac"; import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; import { externalIntakeSchema } from "@/lib/validations/medication"; import { isApiGloballyEnabled } from "@/lib/app-settings"; +import { recomputeMedicationComplianceForEvent } from "@/lib/medications/compliance-rollups"; import { NextRequest, NextResponse } from "next/server"; /** @@ -152,5 +153,19 @@ export const POST = apiHandler(async (request: NextRequest) => { annotate({ meta: { medication_id: medication.id, event_id: event.id } }); + // v1.4.39 W-MED — refresh the compliance rollup for the ingested + // event's user-day. The external ingest path runs without a User + // record in scope; resolve the tz once for the recompute. + const ingestUser = await prisma.user.findUnique({ + where: { id: apiToken.userId }, + select: { timezone: true }, + }); + await recomputeMedicationComplianceForEvent({ + userId: apiToken.userId, + medicationId: medication.id, + scheduledFor: event.scheduledFor, + tz: ingestUser?.timezone ?? null, + }); + return apiSuccess(event, 201); }); diff --git a/src/app/api/medications/[id]/intake/[eventId]/route.ts b/src/app/api/medications/[id]/intake/[eventId]/route.ts index 0214d67cf..8dc51a951 100644 --- a/src/app/api/medications/[id]/intake/[eventId]/route.ts +++ b/src/app/api/medications/[id]/intake/[eventId]/route.ts @@ -10,6 +10,7 @@ import { } from "@/lib/api-response"; import { updateIntakeEventSchema } from "@/lib/validations/medication"; import { invalidateUserMedications } from "@/lib/cache/invalidate"; +import { recomputeMedicationComplianceForEvent } from "@/lib/medications/compliance-rollups"; import { NextRequest } from "next/server"; type RouteParams = { params: Promise<{ id: string; eventId: string }> }; @@ -67,6 +68,28 @@ export const PUT = apiHandler( // caches so the next read reflects the edited event. invalidateUserMedications(user.id); + // v1.4.39 W-MED — refresh the rollup row for both the old + new + // scheduledFor day-keys when the timestamp moved across a day + // boundary; if only takenAt / skipped changed the second call + // collapses to a no-op upsert. + await recomputeMedicationComplianceForEvent({ + userId: user.id, + medicationId: id, + scheduledFor: event.scheduledFor, + tz: user.timezone, + }); + if ( + data.scheduledFor !== undefined && + data.scheduledFor.getTime() !== event.scheduledFor.getTime() + ) { + await recomputeMedicationComplianceForEvent({ + userId: user.id, + medicationId: id, + scheduledFor: data.scheduledFor, + tz: user.timezone, + }); + } + return apiSuccess(updated); }, ); @@ -107,6 +130,16 @@ export const DELETE = apiHandler( // caches so the next read reflects the deletion. invalidateUserMedications(user.id); + // v1.4.39 W-MED — refresh the rollup row for the day the deleted + // event sat on. When that day now holds zero events the helper + // drops the rollup row entirely. + await recomputeMedicationComplianceForEvent({ + userId: user.id, + medicationId: id, + scheduledFor: event.scheduledFor, + tz: user.timezone, + }); + return apiSuccess({ deleted: true }); }, ); diff --git a/src/app/api/medications/[id]/intake/import/route.ts b/src/app/api/medications/[id]/intake/import/route.ts index b4e400651..6dc81501d 100644 --- a/src/app/api/medications/[id]/intake/import/route.ts +++ b/src/app/api/medications/[id]/intake/import/route.ts @@ -8,6 +8,10 @@ import { getClientIp, safeJson, } from "@/lib/api-response"; +import { + recomputeMedicationComplianceForDay, + dayKeyForScheduledFor, +} from "@/lib/medications/compliance-rollups"; import { NextRequest } from "next/server"; import { z } from "zod/v4"; @@ -51,6 +55,9 @@ export const POST = apiHandler( let imported = 0; let skippedDuplicates = 0; let skippedInvalid = 0; + // v1.4.39 W-MED — collect distinct day-keys touched by the import + // so the rollup pass fires once per day rather than once per row. + const touchedDays = new Set(); for (const entry of entries) { // Parse as local datetime first; if invalid, fall back to CET offset. @@ -88,6 +95,30 @@ export const POST = apiHandler( }, }); imported++; + touchedDays.add(dayKeyForScheduledFor(takenAt, user.timezone)); + } + + // v1.4.39 W-MED — fold the touched days into rollup rows after the + // insert loop completes. Best-effort: a populator failure logs but + // never rolls back the imported events. + for (const dayKey of touchedDays) { + try { + await recomputeMedicationComplianceForDay( + user.id, + id, + dayKey, + user.timezone, + ); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + annotate({ + meta: { + medication_compliance_rollup_import_failed: true, + medication_compliance_rollup_import_error: message, + medication_compliance_rollup_day: dayKey, + }, + }); + } } await auditLog("medication.intake.import", { diff --git a/src/app/api/medications/[id]/intake/purge/route.ts b/src/app/api/medications/[id]/intake/purge/route.ts index a29e5ce6d..411d83080 100644 --- a/src/app/api/medications/[id]/intake/purge/route.ts +++ b/src/app/api/medications/[id]/intake/purge/route.ts @@ -21,6 +21,13 @@ export const DELETE = apiHandler( where: { medicationId: id, userId: user.id }, }); + // v1.4.39 W-MED — drop every rollup row for this medication so the + // next compliance read returns zero-filled buckets rather than the + // last-known scheduled / taken totals for now-deleted events. + await prisma.medicationComplianceRollup.deleteMany({ + where: { medicationId: id, userId: user.id }, + }); + await auditLog("medication.intake.purge", { userId: user.id, ipAddress: getClientIp(request), diff --git a/src/app/api/medications/[id]/intake/route.ts b/src/app/api/medications/[id]/intake/route.ts index 3c0fb1c65..9f3ed94b9 100644 --- a/src/app/api/medications/[id]/intake/route.ts +++ b/src/app/api/medications/[id]/intake/route.ts @@ -16,6 +16,7 @@ import { withIdempotency } from "@/lib/idempotency"; import { consumeOneDose } from "@/lib/medications/inventory/service"; import { assertMedicationOwnership } from "@/lib/medications/route-guards"; import { invalidateUserMedications } from "@/lib/cache/invalidate"; +import { recomputeMedicationComplianceForEvent } from "@/lib/medications/compliance-rollups"; import { NextRequest } from "next/server"; type RouteParams = { params: Promise<{ id: string }> }; @@ -159,6 +160,16 @@ async function postIntake(request: NextRequest, { params }: RouteParams) { // caches so the next read reflects the dose event. invalidateUserMedications(user.id); + // v1.4.39 W-MED — refresh the persistent compliance rollup for the + // affected day. The hook is best-effort; failures annotate but never + // block the user's POST response. + await recomputeMedicationComplianceForEvent({ + userId: user.id, + medicationId: id, + scheduledFor: event.scheduledFor, + tz: user.timezone, + }); + return apiSuccess(event, 201); } diff --git a/src/app/api/medications/intake/bulk/route.ts b/src/app/api/medications/intake/bulk/route.ts index fb7aa846b..97fafa880 100644 --- a/src/app/api/medications/intake/bulk/route.ts +++ b/src/app/api/medications/intake/bulk/route.ts @@ -41,6 +41,10 @@ import { import { withIdempotency } from "@/lib/idempotency"; import { checkRateLimit } from "@/lib/rate-limit"; import { invalidateUserMedications } from "@/lib/cache/invalidate"; +import { + recomputeMedicationComplianceForDay, + dayKeyForScheduledFor, +} from "@/lib/medications/compliance-rollups"; const MAX_ENTRIES_PER_BATCH = 500; const BATCH_RATE_LIMIT_MAX = 60; @@ -126,6 +130,11 @@ async function postBulk(request: NextRequest): Promise { let inserted = 0; let duplicates = 0; const skipped: Array<{ index: number; reason: string }> = []; + // v1.4.39 W-MED — collect distinct `(medicationId, dayKey)` pairs + // touched by the batch so one rollup recompute fires per pair after + // all inserts complete. Per-row recompute would balloon a 500-entry + // batch into 500 sequential rollup hits. + const touchedDays = new Map(); for (let i = 0; i < entries.length; i++) { const entry = entries[i]; @@ -155,6 +164,11 @@ async function postBulk(request: NextRequest): Promise { }); inserted += 1; results.push({ index: i, status: "inserted", id: row.id }); + const dayKey = dayKeyForScheduledFor(scheduledFor, user.timezone); + touchedDays.set(`${entry.medicationId}|${dayKey}`, { + medicationId: entry.medicationId, + dayKey, + }); } catch (err: unknown) { // P2002 = unique-constraint violation; the idempotencyKey // already exists → "duplicate", not an error. @@ -213,6 +227,32 @@ async function postBulk(request: NextRequest): Promise { invalidateUserMedications(user.id); } + // v1.4.39 W-MED — refresh one rollup row per distinct + // `(medicationId, dayKey)` touched by the batch. Best-effort: failures + // are swallowed by the helper's `recomputeMedicationComplianceForEvent` + // wrapper, but bulk callers wrap recompute directly to keep the + // tight ingest loop unaffected by populator issues. + for (const { medicationId, dayKey } of touchedDays.values()) { + try { + await recomputeMedicationComplianceForDay( + user.id, + medicationId, + dayKey, + user.timezone, + ); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + annotate({ + meta: { + medication_compliance_rollup_bulk_failed: true, + medication_compliance_rollup_bulk_error: message, + medication_compliance_rollup_medication: medicationId, + medication_compliance_rollup_day: dayKey, + }, + }); + } + } + return apiSuccess({ processed: entries.length, inserted, diff --git a/src/app/api/medications/intake/route.ts b/src/app/api/medications/intake/route.ts index e21d00228..3951e5e65 100644 --- a/src/app/api/medications/intake/route.ts +++ b/src/app/api/medications/intake/route.ts @@ -26,6 +26,13 @@ import { auditLog } from "@/lib/auth/audit"; import { userDayKey, DEFAULT_TIMEZONE } from "@/lib/tz/resolver"; import { cached, caches, type ServerCache } from "@/lib/cache/server-cache"; import { invalidateUserMedications } from "@/lib/cache/invalidate"; +import { + readMedicationCompliance, + hasMedicationComplianceCoverage, + recomputeMedicationComplianceForEvent, + enqueueBootTimeMedicationComplianceBackfill, + type ComplianceBucket as RollupComplianceBucket, +} from "@/lib/medications/compliance-rollups"; const querySchema = z.object({ scope: z.enum(["today", "compliance"]), @@ -116,10 +123,16 @@ export const GET = apiHandler(async (request: NextRequest) => { // are slow-moving (intake events trickle in, yesterday's row doesn't // move). Cache key carries the userTz so a user who changes timezone // doesn't read another tz's bucketing. + // + // v1.4.39 W-MED — read path now consumes the persistent + // `medication_compliance_rollups` tier. A coverage probe falls back + // to the legacy live aggregator when the user has intake events but + // zero rollup rows; the fallback fires a boot-backfill enqueue in the + // background so the next request hits the rollup tier. const result = await cached( caches.medicationsIntake as ServerCache, `${user.id}|compliance|${days}|${userTz}`, - () => buildComplianceBuckets(user.id, days, userTz), + () => readComplianceBucketsWithFallback(user.id, days, userTz), annotate, ); @@ -131,6 +144,44 @@ export const GET = apiHandler(async (request: NextRequest) => { return apiSuccess(result); }); +/** + * v1.4.39 W-MED — read from the persistent rollup tier when covered, + * fall through to the legacy live aggregator on coverage miss. The + * fallback fires the boot backfill in the background so subsequent + * requests hit the rollup path; the current request still returns a + * correct response built from the raw events. + */ +async function readComplianceBucketsWithFallback( + userId: string, + days: number, + userTz: string, +): Promise { + const covered = await hasMedicationComplianceCoverage(userId, days, userTz); + if (covered) { + const buckets = await readMedicationCompliance(userId, days, userTz); + annotate({ + meta: { + medication_compliance_path: "rollup", + medication_compliance_days: days, + }, + }); + return buckets satisfies RollupComplianceBucket[]; + } + + // Coverage miss — the legacy aggregator stays correct over an empty + // rollup window. Fire the boot backfill in the background so the + // next request lands on the rollup tier; the current request still + // returns the live-derived buckets. + void enqueueBootTimeMedicationComplianceBackfill(); + annotate({ + meta: { + medication_compliance_path: "live-fallback", + medication_compliance_days: days, + }, + }); + return buildComplianceBuckets(userId, days, userTz); +} + interface ComplianceBucket { date: string; scheduled: number; @@ -194,6 +245,8 @@ export const POST = apiHandler(async (request: NextRequest) => { return apiError("Intake event not found", 404); } + const userTzForHook = user.timezone ?? DEFAULT_TIMEZONE; + let updated; if (status === "taken") { [updated] = await prisma.$transaction([ @@ -238,5 +291,16 @@ export const POST = apiHandler(async (request: NextRequest) => { // caches for this user so the next read reflects the dose change. invalidateUserMedications(user.id); + // v1.4.39 W-MED — refresh the persistent compliance rollup row for + // the affected day so the next read after the cache miss returns + // the up-to-date `(scheduled, taken, skipped)` tuple. Best-effort: + // a populator failure annotates ops without blocking the response. + await recomputeMedicationComplianceForEvent({ + userId: user.id, + medicationId: existing.medicationId, + scheduledFor: existing.scheduledFor, + tz: userTzForHook, + }); + return apiSuccess(updated); }); diff --git a/src/app/api/telegram/webhook/route.ts b/src/app/api/telegram/webhook/route.ts index ad38d209f..37bab4d16 100644 --- a/src/app/api/telegram/webhook/route.ts +++ b/src/app/api/telegram/webhook/route.ts @@ -13,6 +13,7 @@ import { apiHandler } from "@/lib/api-handler"; import { annotate, getEvent } from "@/lib/logging/context"; import { getServerTranslator } from "@/lib/i18n/server-translator"; import { locales, type Locale } from "@/lib/i18n/config"; +import { recomputeMedicationComplianceForEvent } from "@/lib/medications/compliance-rollups"; interface TelegramUpdate { update_id: number; @@ -115,6 +116,9 @@ async function findTelegramUser(chatId: string) { id: true, telegramBotToken: true, locale: true, + // v1.4.39 W-MED — pulled into the select so the post-write + // compliance-rollup hook can anchor the user-day bucket. + timezone: true, }, }); } @@ -124,6 +128,7 @@ async function markMedicationTaken( medicationId: string, idempotencyKey: string, locale: Locale, + userTz: string | null, ): Promise<{ ok: boolean; message: string; medicationName?: string }> { const { t } = getServerTranslator(locale); const medication = await prisma.medication.findFirst({ @@ -139,14 +144,16 @@ async function markMedicationTaken( select: { id: true }, }); + let scheduledFor: Date | null = null; if (!existing) { + scheduledFor = new Date(); await prisma.$transaction([ prisma.medicationIntakeEvent.create({ data: { userId, medicationId: medication.id, - scheduledFor: new Date(), - takenAt: new Date(), + scheduledFor, + takenAt: scheduledFor, skipped: false, source: "REMINDER", idempotencyKey, @@ -159,6 +166,18 @@ async function markMedicationTaken( ]); } + // v1.4.39 W-MED — refresh the compliance rollup for the + // affected day so the next read after the cache miss reflects the + // new dose. Only fires when the create-path landed a new row. + if (scheduledFor) { + await recomputeMedicationComplianceForEvent({ + userId, + medicationId: medication.id, + scheduledFor, + tz: userTz, + }); + } + return { ok: true, message: existing @@ -205,6 +224,7 @@ async function handleCallback(update: TelegramUpdate) { medicationId, idempotencyKey, locale, + user.timezone, ); await answerTelegramCallbackQuery(botToken, callback.id, result.message); if (messageId) { @@ -289,16 +309,18 @@ async function handleCallback(update: TelegramUpdate) { select: { id: true }, }); + let skippedScheduledFor: Date | null = null; if (!existing) { const endOfDay = new Date(); endOfDay.setHours(23, 59, 59, 999); + skippedScheduledFor = new Date(); await prisma.$transaction([ prisma.medicationIntakeEvent.create({ data: { userId: user.id, medicationId: medication.id, - scheduledFor: new Date(), + scheduledFor: skippedScheduledFor, takenAt: null, skipped: true, source: "REMINDER", @@ -312,6 +334,18 @@ async function handleCallback(update: TelegramUpdate) { ]); } + // v1.4.39 W-MED — refresh the compliance rollup for the skipped + // event's user-day so the next read sees the (scheduled, skipped) + // counts incremented. + if (skippedScheduledFor) { + await recomputeMedicationComplianceForEvent({ + userId: user.id, + medicationId: medication.id, + scheduledFor: skippedScheduledFor, + tz: user.timezone, + }); + } + await answerTelegramCallbackQuery( botToken, callback.id, @@ -372,6 +406,7 @@ async function handleCallback(update: TelegramUpdate) { medicationId, idempotencyKey, locale, + user.timezone, ); await answerTelegramCallbackQuery(botToken, callback.id, result.message); if (messageId) { @@ -477,6 +512,7 @@ async function handleTextMessage(update: TelegramUpdate) { meds[0].id, idempotencyKey, locale, + user.timezone, ); const resp = await sendTelegramMessage( botToken, @@ -611,6 +647,7 @@ async function handleTextMessage(update: TelegramUpdate) { medicationId, idempotencyKey, locale, + user.timezone, ); const resp = await sendTelegramMessage( botToken,