perf(medications-intake): consume compliance rollup tier

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.
This commit is contained in:
Marc-André Bombeck committed 2026-05-18 12:30:25 +02:00
1 parent 83d96a9817
commit 227eb69b63
9 files changed
+246 -4

No files matched your search

@@ -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";
+15
View File
@@ -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);
});
@@ -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 });
},
);
@@ -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<string>();
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", {
@@ -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),
@@ -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);
}
@@ -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<Response> {
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<string, { medicationId: string; dayKey: string }>();
for (let i = 0; i < entries.length; i++) {
const entry = entries[i];
@@ -155,6 +164,11 @@ async function postBulk(request: NextRequest): Promise<Response> {
});
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<Response> {
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,
+65 -1
View File
@@ -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<ComplianceBucket[]>,
`${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<ComplianceBucket[]> {
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);
});
+40 -3
View File
@@ -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,