feat(api): batch routes enqueue pr-detection (silent on >50 batch)

Measurement and workout batch ingest now fire a pr-detection job
after each successful insert. Batches above 50 entries propagate
silent=true so a multi-year Apple Health or HKWorkout backfill
writes the records without spamming hundreds of pushes. Both
enqueue calls are wrapped in audit-log entries and a swallow-and-
annotate failure path so a flaky queue cannot 500 an ingest call —
the 30-minute fallback cron picks the user up in any case.
This commit is contained in:
Marc-André Bombeck committed 2026-05-14 18:47:22 +02:00
1 parent d34c30eab6
commit 7b7a896c44
4 files changed
+240

No files matched your search

@@ -0,0 +1,119 @@
/**
* v1.4.25 W16c — PR detection enqueue coverage for
* `POST /api/measurements/batch`. The full per-entry envelope is
* exercised by the integration suite; this file checks the hook fires
* with the right `silent` flag for the small / large batch cases and
* stays out of the way when nothing landed.
*/
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";
import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
user: { id: "user-1", username: "marc", 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),
});
}
function validStepEntry(externalId: string) {
return {
hkIdentifier: "HKQuantityTypeIdentifierStepCount",
value: 1200,
unit: "count",
startDate: "2026-05-14T06:30:00.000Z",
endDate: "2026-05-14T06:45:00.000Z",
externalId,
};
}
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: 0 });
});
describe("POST /api/measurements/batch — PR detection enqueue (v1.4.25 W16c)", () => {
it("enqueues with silent=false for a small healthy batch", async () => {
vi.mocked(prisma.measurement.createMany).mockResolvedValue({ count: 1 });
const res = await POST(makeRequest({ entries: [validStepEntry("ext-a")] }));
expect(res.status).toBe(200);
expect(enqueuePrDetection).toHaveBeenCalledTimes(1);
expect(enqueuePrDetection).toHaveBeenCalledWith("user-1", {
silent: false,
});
});
it("sets silent=true once the batch crosses the 50-entry threshold", async () => {
const big = Array.from({ length: 51 }, (_, i) =>
validStepEntry(`ext-${i}`),
);
vi.mocked(prisma.measurement.createMany).mockResolvedValue({ count: 51 });
const res = await POST(makeRequest({ entries: big }));
expect(res.status).toBe(200);
expect(enqueuePrDetection).toHaveBeenCalledWith("user-1", {
silent: true,
});
});
});
+37
View File
@@ -36,8 +36,16 @@ import { mapAppleHealthEntry } from "@/lib/measurements/apple-health-mapping";
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 { Prisma } from "@/generated/prisma/client";
// v1.4.25 W16c — historical-backfill threshold for PR push
// suppression. A batch larger than this fires the detection job with
// `silent: true` so a multi-year Apple Health backfill writes records
// without spamming the user with hundreds of pushes. Tuned generously
// — a healthy daily-sync batch is typically well under 50 entries.
const PR_DETECTION_SILENT_THRESHOLD = 50;
const MAX_BATCH_ENTRIES = 500;
// v1.4.25 W10 reconcile (security H-2): cap batch ingest at 60
@@ -319,6 +327,35 @@ async function postBatch(request: NextRequest): Promise<Response> {
},
});
// v1.4.25 W16c — kick off PR detection for this user. We always
// enqueue when at least one row was written (or the batch had any
// measurements to consider) so a single off-day reading still gets
// evaluated; the warm-up gate inside the detector decides whether
// it's a record. Suppress push notifications for historical
// backfills above the silent threshold.
if (insertedCount > 0 || duplicateCount > 0) {
const silent = entries.length > PR_DETECTION_SILENT_THRESHOLD;
try {
await enqueuePrDetection(user.id, { silent });
await auditLog("personal_records.detection_enqueued", {
userId: user.id,
details: {
source: "measurement.batch",
batchSize: entries.length,
silent,
},
});
} catch (err) {
// Enqueue failure is non-fatal — the 30-minute fallback cron
// picks the user up in the next slot. Log so the operator
// notices repeated failures in Wide-Event traffic.
annotate({
action: { name: "personal_records.detection_enqueue_failed" },
meta: { error: err instanceof Error ? err.message : String(err) },
});
}
}
annotate({
action: { name: "measurement.batch.ingest" },
meta: {
@@ -49,6 +49,10 @@ 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 () => ({
@@ -62,6 +66,7 @@ import { POST } from "../batch/route";
import { prisma } from "@/lib/db";
import { getSession } from "@/lib/auth/session";
import { checkRateLimit } from "@/lib/rate-limit";
import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
@@ -349,3 +354,48 @@ describe("POST /api/workouts/batch — nested route attachment", () => {
expect(rows[0]?.workoutId).toBe("wkt-fresh-id");
});
});
describe("POST /api/workouts/batch — PR detection enqueue (v1.4.25 W16c)", () => {
it("enqueues PR detection after a successful batch with silent=false", async () => {
vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 1 });
const res = await POST(
makeRequest({ workouts: [validWorkout("uuid-small")] }),
);
expect(res.status).toBe(200);
expect(enqueuePrDetection).toHaveBeenCalledTimes(1);
expect(enqueuePrDetection).toHaveBeenCalledWith("user-1", {
silent: false,
});
});
it("propagates silent=true when the batch exceeds the historical threshold", async () => {
const big = Array.from({ length: 51 }, (_, i) =>
validWorkout(`uuid-${i}`),
);
vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 51 });
const res = await POST(makeRequest({ workouts: big }));
expect(res.status).toBe(200);
expect(enqueuePrDetection).toHaveBeenCalledWith("user-1", {
silent: true,
});
});
it("still enqueues for an all-duplicate batch (cursor cleanup path)", async () => {
// Pre-flight dedup matches all incoming entries, so insertedCount
// stays 0 but duplicateCount becomes 1. The hook fires anyway —
// the detector is cheap and a duplicate batch can still reveal a
// PR if the row was written by another path between dispatches.
vi.mocked(prisma.workout.findMany).mockResolvedValue([
{ source: "APPLE_HEALTH", externalId: "uuid-dup" } as never,
]);
vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 0 });
const res = await POST(
makeRequest({ workouts: [validWorkout("uuid-dup")] }),
);
expect(res.status).toBe(200);
expect(enqueuePrDetection).toHaveBeenCalledTimes(1);
expect(enqueuePrDetection).toHaveBeenCalledWith("user-1", {
silent: false,
});
});
});
+34
View File
@@ -62,12 +62,20 @@ import {
} from "@/lib/api-response";
import { withIdempotency } from "@/lib/idempotency";
import { checkRateLimit } from "@/lib/rate-limit";
import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
import {
createBatchWorkoutSchema,
MAX_WORKOUTS_PER_BATCH,
} from "@/lib/validations/workout";
import { Prisma } from "@/generated/prisma/client";
// v1.4.25 W16c — push-suppression threshold for workout PRs. A batch
// larger than this fires the detection job with `silent: true` so a
// multi-year HKWorkout backfill doesn't spam the user with hundreds of
// pushes. Matches the measurements-batch threshold so behaviour stays
// uniform across both ingest paths.
const PR_DETECTION_SILENT_THRESHOLD = 50;
// v1.4.25 W10 reconcile (security H-2 parity): cap batch ingest at 60
// batches per user per minute. Healthy iOS sync drains its HealthKit
// observer queue in well under one batch per minute (the observer
@@ -449,6 +457,32 @@ async function postBatch(request: NextRequest): Promise<Response> {
},
});
// v1.4.25 W16c — kick off PR detection for this user. Suppress push
// notifications for historical-backfill batches so a multi-year
// HKWorkout import doesn't fire hundreds of pushes during initial
// sync. The detector also scans Measurement history on the same
// pass, so a workout-only batch still surfaces measurement-side PRs
// the user may have racked up since the last detection.
if (insertedCount > 0 || duplicateCount > 0) {
const silent = workouts.length > PR_DETECTION_SILENT_THRESHOLD;
try {
await enqueuePrDetection(user.id, { silent });
await auditLog("personal_records.detection_enqueued", {
userId: user.id,
details: {
source: "workout.batch",
batchSize: workouts.length,
silent,
},
});
} catch (err) {
annotate({
action: { name: "personal_records.detection_enqueue_failed" },
meta: { error: err instanceof Error ? err.message : String(err) },
});
}
}
annotate({
action: { name: "workout.batch.ingest" },
meta: {