diff --git a/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts b/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts new file mode 100644 index 000000000..35300709e --- /dev/null +++ b/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts @@ -0,0 +1,269 @@ +/** + * v1.32.1 (issue #585) — wide-event observability for the go-forward + * aggregated 10-min heart-rate bucket wire contract. + * + * The intraday pulse chart reads exactly the `stats:HKQuantityTypeIdentifier + * HeartRate:<10-min-Z>` rows this batch route accepts (or rejects). A report + * of "the chart looks sparser after the ZIP-import cutoff" was previously + * undiagnosable from server logs: the baseline `measurement.batch.ingest` + * annotation carried only a total `skipped` count with no reason breakdown + * and no visibility into how many of a batch's entries specifically targeted + * an HR bucket. These tests pin the new `measurement.batch.hr-bucket` + * annotation (fires only when the batch actually carried an HR-bucket entry) + * and the `skipped_by_reason` breakdown on the baseline annotation. + */ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import { NextRequest } from "next/server"; + +const mocks = vi.hoisted(() => ({ annotate: vi.fn() })); + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + update: vi.fn(), + }, + measurement: { + findMany: vi.fn(), + createManyAndReturn: vi.fn(), + updateMany: 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/logging/context", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, annotate: mocks.annotate }; +}); + +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: async ( + tx: { + measurement: { + findMany: (args: unknown) => Promise< + Array<{ + id?: string; + type?: string; + source?: string; + externalId?: string | null; + }> + >; + createManyAndReturn: (args: { + data: Record; + }) => Promise>>; + updateMany: (args: { + where: Record; + data: Record; + }) => Promise; + }; + }, + input: Record & { + userId: string; + type: string; + source: string; + externalId: string; + }, + options: { exactExternalMatch?: "update" | "duplicate" } = {}, + ) => { + const existing = await tx.measurement.findMany({}); + const exact = existing.find( + (row: { type?: string; source?: string; externalId?: string | null }) => + row.type === input.type && + row.source === input.source && + row.externalId === input.externalId, + ); + if (exact && options.exactExternalMatch !== "update") { + return { status: "duplicate", row: exact }; + } + if (exact) { + await tx.measurement.updateMany({ + where: { id: exact.id }, + data: { ...input, deletedAt: null }, + }); + return { status: "updated", row: { ...exact, ...input } }; + } + const [inserted] = await tx.measurement.createManyAndReturn({ + data: input, + }); + if (inserted) return { status: "inserted", row: inserted }; + if (options.exactExternalMatch === "update") { + await tx.measurement.updateMany({ where: {}, data: input }); + return { status: "updated", row: input }; + } + return { status: "duplicate" }; + }, +})); + +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("@/lib/jobs/reminder-satisfy", () => ({ + enqueueReminderSatisfy: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/cache/invalidate", () => ({ + invalidateUserMeasurements: vi.fn(), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + collapseToTypeDayKeys: vi.fn(() => []), +})); + +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 }, +}; + +const FRESH_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:00:00.000Z"; +const REPOST_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T15:00:00.000Z"; +const MALFORMED_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:35:00.000Z"; +const SAMPLE_ID = "C0FFEE00-0000-4000-8000-000000000000"; + +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 hrEntry(externalId: string, value: number) { + return { + hkIdentifier: "HKQuantityTypeIdentifierHeartRate", + value, + unit: "count/min", + startDate: "2026-06-21T14:00:00.000Z", + endDate: "2026-06-21T14:59:59.000Z", + externalId, + }; +} + +function annotateCallsFor(actionName: string) { + return mocks.annotate.mock.calls.filter( + ([arg]) => + (arg as { action?: { name?: string } })?.action?.name === actionName, + ); +} + +beforeEach(() => { + vi.clearAllMocks(); + 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.createManyAndReturn).mockImplementation((async ( + args: unknown, + ) => { + const { data } = args as { + data: + | { type: unknown; source: string; externalId: string | null } + | Array<{ type: unknown; source: string; externalId: string | null }>; + }; + const rows = Array.isArray(data) ? data : [data]; + return rows.map(({ type, source, externalId }) => ({ + type, + source, + externalId, + })); + }) as never); + vi.mocked(prisma.measurement.updateMany).mockResolvedValue({ count: 1 }); +}); + +describe("POST /api/measurements/batch — HR-bucket observability (issue #585)", () => { + it("reports a per-outcome breakdown covering only HR-bucket entries", async () => { + // REPOST_BUCKET_ID already exists — the reconcile mock overwrites it. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + type: "PULSE", + source: "APPLE_HEALTH", + externalId: REPOST_BUCKET_ID, + }, + ] as never); + + const res = await POST( + makeRequest({ + entries: [ + hrEntry(FRESH_BUCKET_ID, 72), // inserted + hrEntry(REPOST_BUCKET_ID, 78), // updated (overwrite) + hrEntry(MALFORMED_BUCKET_ID, 70), // skipped (malformed) + hrEntry(SAMPLE_ID, 64), // NOT an HR bucket — must be excluded + ], + }), + ); + expect(res.status).toBe(200); + + const calls = annotateCallsFor("measurement.batch.hr-bucket"); + expect(calls).toHaveLength(1); + const meta = (calls[0][0] as { meta: Record }).meta; + expect(meta).toEqual({ + attempted: 3, + inserted: 1, + updated: 1, + duplicate: 0, + skipped: 1, + failed: 0, + processed: 4, + }); + }); + + it("never fires the HR-bucket annotation for a batch with no bucket entries", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + const res = await POST(makeRequest({ entries: [hrEntry(SAMPLE_ID, 64)] })); + expect(res.status).toBe(200); + expect(annotateCallsFor("measurement.batch.hr-bucket")).toHaveLength(0); + }); + + it("breaks the baseline ingest annotation's skip count down by reason", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + const res = await POST( + makeRequest({ + entries: [ + hrEntry(MALFORMED_BUCKET_ID, 70), // skipped: malformed_hr_bucket_id + ], + }), + ); + expect(res.status).toBe(200); + + const calls = annotateCallsFor("measurement.batch.ingest"); + expect(calls).toHaveLength(1); + const meta = calls[0][0] as { meta: { skipped_by_reason: unknown } }; + expect(meta.meta.skipped_by_reason).toEqual({ + malformed_hr_bucket_id: 1, + }); + }); +}); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index b739ed6b4..a48cf1c66 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -603,6 +603,18 @@ async function postBatch(request: NextRequest): Promise { .filter((r) => r.status === "skipped") .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); + // v1.32.1 (issue #585) — per-reason skip breakdown. The baseline + // annotation below only ever carried the total `skipped` count, so an + // operator investigating "why does this account's uploaded data look + // sparser than expected" had no way to see WHICH validation gate an + // entry tripped (malformed bucket id, out-of-range value, unmappable + // identifier, …) without pulling raw rows. Grouping by reason turns + // that into a one-glance wide-event field. + const skippedByReason: Record = {}; + for (const s of skipped) { + skippedByReason[s.reason] = (skippedByReason[s.reason] ?? 0) + 1; + } + await auditLog("measurement.batch.ingest", { userId: user.id, ipAddress: getClientIp(request), @@ -648,6 +660,64 @@ async function postBatch(request: NextRequest): Promise { }); } + // v1.32.1 (issue #585) — dedicated outcome breakdown for entries + // targeting the go-forward aggregated 10-min heart-rate bucket prefix + // (`stats:HKQuantityTypeIdentifierHeartRate:<10-min-Z>`). The intraday + // pulse chart reads exactly these rows for a live-synced day; a report + // of "the chart looks sparser after the ZIP import cutoff" is + // unanswerable from the baseline ingest trace alone, since it never + // distinguished an HR-bucket write from any other entry in the batch. + // Grepping `measurement.batch.hr-bucket` now shows, per sync, how many + // bucket entries the client sent and what happened to each — confirming + // (or ruling out) server-side rejection/dedup as the density-gap cause. + // Only fires when the batch actually carried at least one HR-bucket + // entry so the baseline trace is unchanged for every other sync. + let hrBucketInserted = 0; + let hrBucketUpdated = 0; + let hrBucketDuplicate = 0; + let hrBucketSkipped = 0; + let hrBucketFailed = 0; + for (let index = 0; index < entries.length; index++) { + if (!targetsAggregatedBucket(entries[index].externalId)) continue; + switch (results[index]?.status) { + case "inserted": + hrBucketInserted++; + break; + case "updated": + hrBucketUpdated++; + break; + case "duplicate": + hrBucketDuplicate++; + break; + case "skipped": + hrBucketSkipped++; + break; + case "failed": + hrBucketFailed++; + break; + } + } + const hrBucketAttempted = + hrBucketInserted + + hrBucketUpdated + + hrBucketDuplicate + + hrBucketSkipped + + hrBucketFailed; + if (hrBucketAttempted > 0) { + annotate({ + action: { name: "measurement.batch.hr-bucket" }, + meta: { + attempted: hrBucketAttempted, + inserted: hrBucketInserted, + updated: hrBucketUpdated, + duplicate: hrBucketDuplicate, + skipped: hrBucketSkipped, + failed: hrBucketFailed, + processed: entries.length, + }, + }); + } + // 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 @@ -693,6 +763,9 @@ async function postBatch(request: NextRequest): Promise { updated: updatedCount, duplicates: duplicateCount, skipped: skipped.length, + // v1.32.1 (issue #585) — per-reason skip breakdown; see the + // `skippedByReason` build above. + skipped_by_reason: skippedByReason, failed: failedCount, }, }); diff --git a/src/lib/analytics/__tests__/intraday-pulse-io.test.ts b/src/lib/analytics/__tests__/intraday-pulse-io.test.ts index 867c0b48c..57450a5b4 100644 --- a/src/lib/analytics/__tests__/intraday-pulse-io.test.ts +++ b/src/lib/analytics/__tests__/intraday-pulse-io.test.ts @@ -13,6 +13,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; const findManyMock = vi.fn(); const workoutFindManyMock = vi.fn(); +const { annotateMock } = vi.hoisted(() => ({ annotateMock: vi.fn() })); vi.mock("@/lib/db", () => ({ prisma: { @@ -21,6 +22,8 @@ vi.mock("@/lib/db", () => ({ }, })); +vi.mock("@/lib/logging/context", () => ({ annotate: annotateMock })); + import { loadIntradayPulse } from "@/lib/analytics/intraday-pulse-io"; import { resolveRestingPulseSeries, @@ -42,6 +45,7 @@ describe("loadIntradayPulse", () => { findManyMock.mockReset(); workoutFindManyMock.mockReset(); workoutFindManyMock.mockResolvedValue([]); + annotateMock.mockReset(); }); it("M2 — queries PULSE/step rows against the exact local-day window, not the padded lead/trail superset", async () => { @@ -209,4 +213,59 @@ describe("loadIntradayPulse", () => { { startMinute: 490, mean: 75, count: 2, min: 70, max: 81 }, ]); }); + + it("v1.32.1 (issue #585) — annotates the per-shape row/bucket breakdown for observability", async () => { + const tz = "UTC"; + const dateKey = "2026-06-15"; + + const restingRows = Array.from({ length: 20 }, (_, i) => ({ + value: 55, + measuredAt: new Date( + `2026-05-${String((i % 28) + 1).padStart(2, "0")}T06:00:00.000Z`, + ), + })); + // One uploaded 10-min bucket row and one raw per-sample row sharing the + // day — the DTO merges them (bucket overlays raw on a slot collision), + // but the annotation must still report each SHAPE's own row count, not + // just the merged bucket total. + const uploadedBucket = { + value: 72, + valueMin: 64, + valueMax: 88, + measuredAt: new Date("2026-06-15T08:09:00.000Z"), + externalId: + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-15T08:00:00.000Z", + }; + const rawSample = { + value: 65, + valueMin: null, + valueMax: null, + measuredAt: new Date("2026-06-15T09:05:00.000Z"), + externalId: "sample:abc123", + }; + + findManyMock.mockImplementation( + async (args: { + where: { type: string; measuredAt?: unknown }; + take?: number; + }) => { + if (args.where.type === "RESTING_HEART_RATE") return restingRows; + if (args.where.type === "PULSE" && args.where.measuredAt) { + return [uploadedBucket, rawSample]; + } + return []; + }, + ); + + await loadIntradayPulse("user-1", tz, dateKey); + + expect(annotateMock).toHaveBeenCalledWith({ + meta: { + intraday_raw_sample_rows: 1, + intraday_uploaded_bucket_rows: 1, + intraday_folded_hourly_rows: 0, + intraday_ten_min_buckets: 2, + }, + }); + }); }); diff --git a/src/lib/analytics/intraday-pulse-io.ts b/src/lib/analytics/intraday-pulse-io.ts index 9a4df4243..ec7987b34 100644 --- a/src/lib/analytics/intraday-pulse-io.ts +++ b/src/lib/analytics/intraday-pulse-io.ts @@ -16,6 +16,7 @@ * digest's `tension_window` builder, so the signal is computed one way. */ import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; import { percentile } from "@/lib/insights/strain-score"; import { resolveRestingPulseSeries } from "@/lib/analytics/resting-pulse"; import { localDayWindow } from "@/lib/measurements/consolidation-tz"; @@ -349,6 +350,26 @@ export async function loadIntradayPulse( }) : null; + // v1.32.1 (issue #585) — per-shape row/bucket breakdown. A report of + // "the chart looks sparser after the ZIP-import cutoff" is otherwise + // undiagnosable from server logs: the route's own annotation only ever + // surfaced the FINAL bucket count, with no visibility into which + // ingestion shape (raw ZIP-imported samples vs uploaded 10-min + // aggregates) the day's data actually came from, nor how many DB rows + // fed each shape before bucketing. Grepping `insights.pulse.intraday` + // now shows both — e.g. `uploaded_bucket_rows: 6` next to + // `ten_min_buckets: 6` confirms the uploaded shape IS being read and + // placed 1:1, narrowing a future report to "the client uploaded too few + // buckets" rather than leaving the read path itself as a live suspect. + annotate({ + meta: { + intraday_raw_sample_rows: rawPulseRows.length, + intraday_uploaded_bucket_rows: uploadedBucketRows.length, + intraday_folded_hourly_rows: foldedPulseRows.length, + intraday_ten_min_buckets: tenMinSeries.length, + }, + }); + return { dateKey, timezone,