fix(compliance,ingest): bound the point window and the two unbounded write paths

`classifyIntakeTiming` treated `windowStart === windowEnd` as an overnight
wrap, inflating a degenerate point window into a ~30h on-time band — a dose
taken many hours late still classified `on_time` and fed a perfect-compliance
streak. Widen the point window symmetrically instead, matching the treatment
`resolveLegacyWindowBounds` already applies.

`/api/measurements/batch` never applied the entry-instant plausibility bound
its single-entry twin has carried since v1.17 W1b, so a future-dated row landed
and permanently won 'latest reading' while poisoning the rollup buckets and the
canonical picker. Skip per entry, like the range guard. The past floor stays at
1900 — a historical backfill is a legitimate shape here.

The intake CSV import had neither an upper date bound nor a rate limiter, while
running up to 1000 sequential find + create + consume iterations per call. Add
both, on the bulk twin's bucket shape and budget.
This commit is contained in:
Marc-André Bombeck committed 2026-07-19 13:26:06 +02:00
1 parent ef10ac5581
commit f5889affb7
6 files changed
+336 -3

No files matched your search

@@ -0,0 +1,158 @@
/**
* `POST /api/measurements/batch` — entry-instant plausibility bound.
*
* The single-entry twin has applied `validateEntryInstant` to `measuredAt`
* since v1.17 W1b; the 500-entry batch path did not, so a future-dated row
* landed and permanently won "latest reading" on every dashboard while
* poisoning the rollup buckets and the canonical picker.
*
* The past floor deliberately stays at 1900: a historical Apple Health
* backfill is a legitimate shape on this route. Only the future side is a bug.
*/
import { describe, expect, it, vi, beforeEach } from "vitest";
import { NextRequest } from "next/server";
vi.mock("@/lib/db", () => ({
prisma: {
measurement: {
findMany: vi.fn(),
createMany: vi.fn(),
},
$transaction: vi.fn(async (fn: unknown) => {
if (typeof fn === "function") {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
return (fn as any)(prisma as unknown as { measurement: unknown });
}
}),
},
}));
vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
vi.mock("@/lib/auth/audit", () => ({
auditLog: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
vi.mock("@/lib/db-compat", () => ({
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/rate-limit", () => ({
checkRateLimit: vi.fn(),
}));
vi.mock("@/lib/jobs/pr-detection", () => ({
enqueuePrDetection: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("next/headers", () => ({
headers: vi.fn(async () => ({ get: () => null })),
cookies: vi.fn(async () => ({
get: () => undefined,
set: () => {},
delete: () => {},
})),
}));
import { POST } from "../route";
import { prisma } from "@/lib/db";
import { getSession } from "@/lib/auth/session";
import { checkRateLimit } from "@/lib/rate-limit";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
user: { id: "user-1", username: "testuser", role: "USER" as const },
};
function makeRequest(body: unknown): NextRequest {
return new NextRequest("http://localhost/api/measurements/batch", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify(body),
});
}
/** A step entry anchored on an offset from now, so the test never date-bombs. */
function stepEntryAt(externalId: string, startMs: number, endMs = startMs) {
return {
hkIdentifier: "HKQuantityTypeIdentifierStepCount",
value: 1200,
unit: "count",
startDate: new Date(startMs).toISOString(),
endDate: new Date(endMs).toISOString(),
externalId,
};
}
const DAY_MS = 24 * 60 * 60 * 1000;
async function statusesFor(entries: unknown[]) {
const res = await POST(makeRequest({ entries }));
expect(res.status).toBe(200);
const body = (await res.json()) as {
data: {
entries: Array<{ index: number; status: string; reason?: string }>;
};
};
return body.data.entries;
}
beforeEach(() => {
vi.resetAllMocks();
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
vi.mocked(checkRateLimit).mockResolvedValue({
allowed: true,
remaining: 60,
resetAt: Date.now() + 60_000,
});
vi.mocked(prisma.measurement.findMany).mockResolvedValue([]);
vi.mocked(prisma.measurement.createMany).mockResolvedValue({ count: 1 });
});
describe("POST /api/measurements/batch — entry-instant plausibility", () => {
it("skips a far-future entry instead of persisting it", async () => {
const results = await statusesFor([
stepEntryAt("ext-future", Date.now() + 365 * DAY_MS),
]);
expect(results[0]).toMatchObject({
index: 0,
status: "skipped",
reason: "implausible_timestamp",
});
});
it("skips an entry predating the 1900 floor", async () => {
const results = await statusesFor([
stepEntryAt("ext-ancient", Date.UTC(1899, 11, 31)),
]);
expect(results[0]).toMatchObject({
status: "skipped",
reason: "implausible_timestamp",
});
});
it("accepts a historical backfill entry — the past floor stays at 1900", async () => {
const results = await statusesFor([
stepEntryAt("ext-old", Date.now() - 8 * 365 * DAY_MS),
]);
expect(results[0]!.status).not.toBe("skipped");
});
it("tolerates a client clock a couple of minutes fast", async () => {
const results = await statusesFor([
stepEntryAt("ext-skew", Date.now() + 2 * 60 * 1000),
]);
expect(results[0]!.status).not.toBe("skipped");
});
it("skips only the offending entry and keeps the rest of the batch", async () => {
const results = await statusesFor([
stepEntryAt("ext-ok", Date.now() - DAY_MS),
stepEntryAt("ext-bad", Date.now() + 365 * DAY_MS),
]);
expect(results[0]!.status).not.toBe("skipped");
expect(results[1]).toMatchObject({ reason: "implausible_timestamp" });
});
});
+17
View File
@@ -60,6 +60,7 @@ import {
oppositeMergeSource,
type MergeCandidate,
} from "@/lib/measurements/cross-source-merge";
import { isPlausibleEntryInstant } from "@/lib/validations/entry-instant";
import { validateMeasurementRange } from "@/lib/validations/measurement";
import { deviceTypeEnum } from "@/lib/validations/source-priority";
import { checkRateLimit } from "@/lib/rate-limit";
@@ -292,6 +293,22 @@ async function postBatch(request: NextRequest): Promise<Response> {
continue;
}
// Time-plausibility guard, the bound the single-entry twin already applies
// through `validateEntryInstant` on `createMeasurementSchema.measuredAt`.
// Without it a future-dated row permanently won "latest reading" on every
// dashboard and poisoned the rollup buckets and the canonical picker. The
// past floor stays at 1900 (no `maxAgeMs`) because a historical Apple
// Health backfill is a legitimate, expected shape on this route; only the
// future side is a bug. Skipped per entry, like the range guard above.
if (!isPlausibleEntryInstant(mapped.takenAt)) {
results[index] = {
index,
status: "skipped",
reason: "implausible_timestamp",
};
continue;
}
// v1.30.7 (iOS #34) — go-forward aggregated heart-rate wire
// contract. A `stats:HKQuantityTypeIdentifierHeartRate:<10-min-ISO-Z>`
// row is the 10-minute-average PULSE bucket; it rides the generic
@@ -58,11 +58,16 @@ vi.mock("@/lib/notifications/medication-intake-sync", () => ({
queueMedicationIntakeSync: vi.fn(),
}));
vi.mock("@/lib/rate-limit", () => ({
checkRateLimit: vi.fn(),
}));
import { POST } from "../route";
import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
import { prisma } from "@/lib/db";
import { getSession } from "@/lib/auth/session";
import { consumeForIntake } from "@/lib/medications/inventory/consumption";
import { checkRateLimit } from "@/lib/rate-limit";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
@@ -89,6 +94,11 @@ beforeEach(() => {
vi.resetAllMocks();
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
vi.mocked(prisma.auditLog.create).mockResolvedValue({} as never);
vi.mocked(checkRateLimit).mockResolvedValue({
allowed: true,
remaining: 60,
resetAt: Date.now() + 60_000,
});
vi.mocked(prisma.medication.findUnique).mockResolvedValue({
id: "m1",
userId: "user-1",
@@ -245,3 +255,73 @@ describe("POST /api/medications/[id]/intake/import — inventory consumption", (
expect(consumeForIntake).toHaveBeenCalledTimes(1);
});
});
describe("POST /api/medications/[id]/intake/import — bounds and limiter", () => {
beforeEach(() => {
vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValue([]);
vi.mocked(prisma.medicationIntakeEvent.create).mockImplementation(
(async (args: { data: { takenAt: Date } }) => ({
id: "evt-1",
takenAt: args.data.takenAt,
})) as never,
);
});
// The `\d{4}-\d{2}-\d{2}` regex has no upper bound. A future-dated row used
// to create a "taken" intake ahead of now, which then fed compliance and the
// dose-history ledger. The single and bulk intake twins bound this through
// `boundedTakenAtSchema`.
it("skips a future-dated row instead of importing it", async () => {
const future = new Date(Date.now() + 365 * 24 * 60 * 60 * 1000);
const datum = future.toISOString().slice(0, 10);
const res = await POST(
postReq([{ datum, uhrzeit: "07:00:00" }]),
ROUTE_CTX,
);
expect(res.status).toBe(201);
const body = (await res.json()) as {
data: { imported: number; skippedInvalid: number };
};
expect(body.data.imported).toBe(0);
expect(body.data.skippedInvalid).toBe(1);
expect(prisma.medicationIntakeEvent.create).not.toHaveBeenCalled();
});
it("still imports a historical row — the past side stays unbounded", async () => {
const res = await POST(
postReq([{ datum: "2019-03-04", uhrzeit: "07:00:00" }]),
ROUTE_CTX,
);
expect(res.status).toBe(201);
const body = (await res.json()) as { data: { imported: number } };
expect(body.data.imported).toBe(1);
});
// This was the one intake write path with no limiter, while running up to
// 1000 sequential find + create + inventory-consume iterations per call.
it("refuses with 429 once the import limiter trips", async () => {
vi.mocked(checkRateLimit).mockResolvedValue({
allowed: false,
remaining: 0,
resetAt: Date.now() + 60_000,
});
const res = await POST(
postReq([{ datum: "2026-01-01", uhrzeit: "07:00:00" }]),
ROUTE_CTX,
);
expect(res.status).toBe(429);
expect(prisma.medicationIntakeEvent.create).not.toHaveBeenCalled();
});
it("buckets the limiter per user", async () => {
await POST(
postReq([{ datum: "2026-01-01", uhrzeit: "07:00:00" }]),
ROUTE_CTX,
);
expect(checkRateLimit).toHaveBeenCalledWith(
"medications:intake:import:user-1",
expect.any(Number),
expect.any(Number),
);
});
});
@@ -3,6 +3,7 @@ import { apiHandler, requireAuth } from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import {
apiError,
apiSuccess,
getClientIp,
returnAllZodIssues,
@@ -17,6 +18,8 @@ import {
recomputeMedicationComplianceForDay,
dayKeyForScheduledFor,
} from "@/lib/rollups/medication-compliance-rollups";
import { checkRateLimit } from "@/lib/rate-limit";
import { isPlausibleEntryInstant } from "@/lib/validations/entry-instant";
import { NextRequest } from "next/server";
import { z } from "zod/v4";
@@ -28,6 +31,10 @@ const importEntrySchema = z.object({
const importSchema = z.array(importEntrySchema).min(1).max(1000);
// Mirrors the bulk intake twin's budget.
const IMPORT_RATE_LIMIT_MAX = 60;
const IMPORT_RATE_LIMIT_WINDOW_MS = 60 * 1000;
type RouteParams = { params: Promise<{ id: string }> };
export const POST = apiHandler(
@@ -40,6 +47,18 @@ export const POST = apiHandler(
const guard = await assertMedicationOwnership(id, user.id);
if (guard) return guard;
// This was the one intake write path without a limiter, while running up
// to 1000 sequential find + create + inventory-consume iterations per
// call. Same bucket shape and budget as the bulk intake twin.
const rl = await checkRateLimit(
`medications:intake:import:${user.id}`,
IMPORT_RATE_LIMIT_MAX,
IMPORT_RATE_LIMIT_WINDOW_MS,
);
if (!rl.allowed) {
return apiError("Too many import submissions, try again later", 429);
}
const { data: body, error: jsonError } = await safeJson(request, {
maxBytes: 1024 * 1024,
});
@@ -124,6 +143,17 @@ export const POST = apiHandler(
continue;
}
// The bare `\d{4}-\d{2}-\d{2}` regex above has no upper bound, so a
// `2999-12-31` row used to create a future "taken" intake that then fed
// compliance and the dose-history ledger. The single and bulk intake
// twins bound this through `boundedTakenAtSchema`; this path did not.
// The past side stays open on purpose — importing years of history is
// the whole point of this route — so only the future bound applies.
if (!isPlausibleEntryInstant(takenAt)) {
skippedInvalid++;
continue;
}
// Prefer explicit counter; otherwise use timestamp-based dedup key.
const idempotencyKey = entry.zaehler
? `import-${id}-${String(entry.zaehler)}`
@@ -1234,6 +1234,43 @@ describe("classifyIntakeTiming", () => {
);
});
// A degenerate POINT window (`windowStart === windowEnd`) is a real shape on
// legacy rows. It used to take the overnight branch and inflate into a ~30h
// on-time band, so a dose taken many hours late still counted `on_time` and
// fed a perfect-compliance streak. It now widens symmetrically by the default
// on-time half-width, matching `resolveLegacyWindowBounds`.
describe("degenerate point window", () => {
it('keeps a dose at the point itself "on_time"', () => {
const takenAt = new Date("2025-01-15T07:00:00Z");
expect(
classifyIntakeTiming(takenAt, "07:00", "07:00", scheduledDate),
).toBe("on_time");
});
it('does not classify a dose ~16h late as "on_time"', () => {
const takenAt = new Date("2025-01-15T23:00:00Z");
expect(
classifyIntakeTiming(takenAt, "07:00", "07:00", scheduledDate),
).toBe("very_late");
});
it("does not swallow the whole day into the on-time band", () => {
// 13:00 is 6h past the point — outside the widened band + graces.
const takenAt = new Date("2025-01-15T13:00:00Z");
expect(
classifyIntakeTiming(takenAt, "07:00", "07:00", scheduledDate),
).not.toBe("on_time");
});
it("still wraps a genuine overnight window", () => {
// 00:30 the next day sits inside a 23:00 -> 01:00 window.
const takenAt = new Date("2025-01-16T00:30:00Z");
expect(
classifyIntakeTiming(takenAt, "23:00", "01:00", scheduledDate),
).toBe("on_time");
});
});
it('returns "on_time" when taken within the window', () => {
const takenAt = new Date("2025-01-15T08:30:00Z");
expect(classifyIntakeTiming(takenAt, "08:00", "09:00", scheduledDate)).toBe(
+14 -3
View File
@@ -1,6 +1,7 @@
// Extracted from the former single-file `compliance.ts`. See `../compliance.ts`
// (the barrel) for the module map. Pure move — no logic changes.
import { DOSE_WINDOW_DEFAULTS } from "@/lib/medications/scheduling/dose-window-defaults";
import { localHmAsUtc } from "@/lib/tz/local-day";
export type IntakeTimingClass =
@@ -68,11 +69,21 @@ export function classifyIntakeTiming(
): IntakeTimingClass {
if (takenAt === null) return "missed";
const start = toDateOnDay(windowStart, scheduledDate, options?.tz);
let start = toDateOnDay(windowStart, scheduledDate, options?.tz);
let end = toDateOnDay(windowEnd, scheduledDate, options?.tz);
// Handle overnight windows (e.g. windowStart="23:00", windowEnd="01:00")
if (end <= start) {
// A degenerate POINT window (`windowStart === windowEnd` — a single dose time
// echoed into both fields, a real shape on legacy rows) used to fall into the
// overnight branch below and inflate into a ~30h on-time band, so a dose taken
// many hours late still classified `on_time` and fed a perfect-compliance
// streak. Widen it symmetrically by the default on-time half-width instead,
// matching `resolveLegacyWindowBounds` in `@/lib/medications/window-status`.
if (end.getTime() === start.getTime()) {
const halfMs = DOSE_WINDOW_DEFAULTS.dailyOnTimeMinutes * 60 * 1000;
start = new Date(start.getTime() - halfMs);
end = new Date(end.getTime() + halfMs);
} else if (end < start) {
// Genuine overnight window (e.g. windowStart="23:00", windowEnd="01:00").
end = new Date(end.getTime() + 24 * 60 * 60 * 1000);
}