Merge remote-tracking branch 'origin/fix/apple-health-pulse-cluster' into release/v1.32.1

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<typeof import("@/lib/logging/context")>();
+  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<string, unknown>;
+        }) => Promise<Array<Record<string, unknown>>>;
+        updateMany: (args: {
+          where: Record<string, unknown>;
+          data: Record<string, unknown>;
+        }) => Promise<unknown>;
+      };
+    },
+    input: Record<string, unknown> & {
+      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<string, number> }).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,
+    });
+  });
+});

src/app/api/measurements/batch/route.ts

@@ -603,6 +603,18 @@ async function postBatch(request: NextRequest): Promise<Response> {
     .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<string, number> = {};
+  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<Response> {
     });
   }
 
+  // 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<Response> {
       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,
     },
   });