Merge remote-tracking branch 'origin/work/v131-slice1-arrival-spine' into release/v1.31.0

src/app/api/labs/ocr/commit/route.ts

@@ -25,6 +25,7 @@ import { auditLog } from "@/lib/auth/audit";
 import { prisma } from "@/lib/db";
 import { withIdempotency } from "@/lib/idempotency";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import { resolveOrMintBiomarker } from "@/lib/labs/biomarker-store";
 import { serialiseLabResult } from "@/lib/labs/serialise";
 import {
@@ -214,6 +215,24 @@ async function commitOcrRows(request: NextRequest) {
     fireAndForget(enqueueReminderSatisfy(user.id), {
       action: "reminder.satisfy.enqueue",
     });
+
+    // v1.31.0 — the labs arm of the data-arrival spine. This seam DOES know
+    // its whole panel (the commit builds a per-row `linkable` list), so it
+    // emits once with the newest draw date and the true inserted count rather
+    // than leaning on the singleton key to collapse per-row emits.
+    const newestTakenAt = linkable.reduce<Date | null>(
+      (acc, r) => (!acc || r.takenAt > acc ? r.takenAt : acc),
+      null,
+    );
+    if (newestTakenAt) {
+      void emitDataArrival({
+        userId: user.id,
+        kind: "labs_panel",
+        newestSampleAt: newestTakenAt,
+        insertedCount: inserted.length,
+        source: "ocr",
+      }).catch(() => {});
+    }
   }
 
   return apiSuccess({ inserted, skipped });

src/app/api/labs/route.ts

@@ -14,6 +14,7 @@ import { auditLog } from "@/lib/auth/audit";
 import { prisma } from "@/lib/db";
 import { withIdempotency } from "@/lib/idempotency";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import { resolveOrMintBiomarker } from "@/lib/labs/biomarker-store";
 import {
   type ResolvedBiomarker,
@@ -276,5 +277,18 @@ async function postLabResult(request: NextRequest) {
     action: "reminder.satisfy.enqueue",
   });
 
+  // v1.31.0 — the labs arm of the data-arrival spine. A "panel" has no
+  // first-class entity in the schema (it is a nullable label plus a `takenAt`),
+  // so the day-scoped singleton key IS the panel grouping: pasting twelve
+  // markers from one draw fires twelve emits that collapse into one arrival.
+  void emitDataArrival({
+    userId: user.id,
+    kind: "labs_panel",
+    newestSampleAt: created.takenAt,
+    insertedCount: 1,
+    refId: created.panel ?? undefined,
+    source: "manual",
+  }).catch(() => {});
+
   return apiSuccess(serialiseLabResult(created, biomarker), 201);
 }

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

@@ -66,6 +66,8 @@ import { checkRateLimit } from "@/lib/rate-limit";
 import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
+import { groupRowsByArrivalKind } from "@/lib/arrivals/measurement-kind";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import {
@@ -177,6 +179,39 @@ type BatchEntry = z.infer<typeof batchEntrySchema>;
  */
 type EntryStatus = "inserted" | "updated" | "duplicate" | "skipped";
 
+/**
+ * v1.31.0 — the rows this write actually INSERTED, in the shape the arrival
+ * kind grouper takes.
+ *
+ * Restricting to `status === "inserted"` is what keeps a re-sync silent: an
+ * `updated` row is the `stats:` overwrite contract re-stating a value the
+ * record already had, and a `duplicate` row landed nothing. The per-entry
+ * statuses are reconciled against the insert race upstream, so the count is
+ * accurate; in a rare concurrent-write tie it can be off on WHICH row, which
+ * cannot change the outcome because the day-scoped singleton key collapses the
+ * result either way.
+ */
+function insertedArrivalRows(
+  prepared: ReadonlyArray<{
+    index: number;
+    row: Prisma.MeasurementCreateManyInput;
+  }>,
+  results: ReadonlyArray<{ status: EntryStatus } | undefined>,
+): Array<{ type: MeasurementType; measuredAt: Date }> {
+  const rows: Array<{ type: MeasurementType; measuredAt: Date }> = [];
+  for (const p of prepared) {
+    if (results[p.index]?.status !== "inserted") continue;
+    rows.push({
+      type: p.row.type as MeasurementType,
+      measuredAt:
+        p.row.measuredAt instanceof Date
+          ? p.row.measuredAt
+          : new Date(p.row.measuredAt as string),
+    });
+  }
+  return rows;
+}
+
 // v1.5.0 issue #213 — `stats:*` externalIds carry per-day cumulative
 // totals that the iOS HealthKit observer re-posts as the day progresses.
 // Treating those as immutable duplicates freezes the day's tile at the
@@ -828,6 +863,35 @@ async function postBatch(request: NextRequest): Promise<Response> {
             : new Date(p.row.measuredAt as string),
         ),
     ).catch(() => {});
+
+    // v1.31.0 — the measurement arm of the data-arrival spine.
+    //
+    // ONE emit per kind per request, never one per row. The batch route is the
+    // Apple-Health / iOS ingest path, so it is also the single biggest backfill
+    // path in the product: a ten-year export arrives here in 500-row batches.
+    // Emitting per row would put thousands of classifier calls (and thousands
+    // of dropped-event annotations) on the hot path for no gain, since the
+    // day-scoped singleton key collapses them to one job anyway.
+    //
+    // Blood pressure is deliberately one arrival, not two: it is stored as two
+    // rows (SYS + DIA) sharing a `measuredAt`, and a reader cares about the
+    // reading, not the arms.
+    //
+    // Only rows the write actually INSERTED count — `updatedCount` rows are the
+    // `stats:` overwrite contract re-stating a value the record already had,
+    // which is not news. If nothing was inserted the classifier returns `noop`
+    // and nothing is enqueued.
+    for (const [kind, group] of groupRowsByArrivalKind(
+      insertedArrivalRows(prepared, results),
+    )) {
+      void emitDataArrival({
+        userId: user.id,
+        kind,
+        newestSampleAt: group.newestAt,
+        insertedCount: group.count,
+        source: "batch",
+      }).catch(() => {});
+    }
   }
 
   return apiSuccess({

src/app/api/measurements/route.ts

@@ -32,6 +32,11 @@ import {
 import { withIdempotency } from "@/lib/idempotency";
 import { encryptNote, shapeMeasurementNotes } from "@/lib/crypto/note-cipher";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
+import {
+  ARRIVAL_MEASUREMENT_KIND,
+  groupRowsByArrivalKind,
+} from "@/lib/arrivals/measurement-kind";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { runSafetyFloorCheck } from "@/lib/illness/safety-floor-check";
@@ -846,6 +851,21 @@ async function postMeasurement(request: NextRequest) {
       action: "reminder.satisfy.enqueue",
     });
 
+    // v1.31.0 — the interactive arm of the data-arrival spine. This route is
+    // the combined BP + pulse form, so it lands both BP arms in one
+    // transaction; they are one reading and produce one arrival, which the
+    // kind map already collapses. Every row here is a real insert (the
+    // transaction throws rather than de-duplicating), so the count is exact.
+    for (const [kind, group] of groupRowsByArrivalKind(results)) {
+      void emitDataArrival({
+        userId: user.id,
+        kind,
+        newestSampleAt: group.newestAt,
+        insertedCount: group.count,
+        source: "manual",
+      }).catch(() => {});
+    }
+
     // v1.18.6 — absolute clinical safety-floor check on the just-written
     // readings (confirm-gated, module-gated, never diagnoses). The combined
     // BP form lands both arms here, so the BP floor sees a whole reading.
@@ -998,6 +1018,22 @@ async function postMeasurement(request: NextRequest) {
     action: "reminder.satisfy.enqueue",
   });
 
+  // v1.31.0 — the single-entry arm of the data-arrival spine. One row, so at
+  // most one kind; a type the spine does not track maps to null and emits
+  // nothing.
+  {
+    const kind = ARRIVAL_MEASUREMENT_KIND[measurement.type];
+    if (kind) {
+      void emitDataArrival({
+        userId: user.id,
+        kind,
+        newestSampleAt: measurement.measuredAt,
+        insertedCount: 1,
+        source: "manual",
+      }).catch(() => {});
+    }
+  }
+
   // v1.18.6 — absolute clinical safety-floor check (confirm-gated,
   // module-gated, never diagnoses). A single-entry POST carries only one arm
   // of a BP reading, so the BP floor only fires through the combined-form /

src/app/api/workouts/__tests__/batch-create.test.ts

@@ -524,7 +524,19 @@ describe("POST /api/workouts/batch — user source-priority (v1.4.43 W9)", () =>
       }),
     );
     expect(res.status).toBe(200);
-    expect(prisma.user.findUnique).toHaveBeenCalledTimes(1);
+    // Counted by SELECT shape rather than by raw call count: the route also
+    // resolves the user's timezone and module map through the arrival seam,
+    // which are separate cached lookups. The guard here is that the
+    // source-priority blob is read ONCE PER BATCH and never once per workout —
+    // two workouts must still produce exactly one read.
+    const sourcePriorityReads = vi
+      .mocked(prisma.user.findUnique)
+      .mock.calls.filter(
+        (call) =>
+          (call[0] as { select?: Record<string, unknown> })?.select
+            ?.sourcePriorityJson === true,
+      );
+    expect(sourcePriorityReads).toHaveLength(1);
     expect(prisma.user.findUnique).toHaveBeenCalledWith({
       where: { id: "user-1" },
       select: { sourcePriorityJson: true },

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

@@ -68,6 +68,7 @@ import { withIdempotency } from "@/lib/idempotency";
 import { checkRateLimit } from "@/lib/rate-limit";
 import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
+import { emitDataArrival } from "@/lib/arrivals/emit-shared";
 import {
   createBatchWorkoutSchema,
   MAX_WORKOUTS_PER_BATCH,
@@ -118,6 +119,65 @@ interface EntryResult {
   reason?: string;
 }
 
+/**
+ * v1.31.0 — emit one data arrival per workout this batch actually inserted.
+ *
+ * `createMany` does not return ids, so they are recovered with a single indexed
+ * lookup on the `(userId, source, externalId)` dedup key. Entries without an
+ * external id cannot be looked up and are emitted without a `refId`; the spine
+ * then falls back to a day-scoped key for them, which is an honest degradation
+ * (no external id means no stable per-workout referent to react to anyway).
+ *
+ * Fully best-effort — every failure path returns quietly. A reaction is never
+ * worth failing an ingest that already succeeded.
+ */
+async function emitWorkoutArrivals(
+  userId: string,
+  prepared: ReadonlyArray<{
+    index: number;
+    row: Prisma.WorkoutCreateManyInput;
+    dedupKey: { source: string; externalId: string } | null;
+  }>,
+  results: ReadonlyArray<EntryResult | undefined>,
+): Promise<void> {
+  const inserted = prepared.filter(
+    (p) => results[p.index]?.status === "inserted",
+  );
+  if (inserted.length === 0) return;
+
+  const externalIds = inserted
+    .map((p) => p.dedupKey?.externalId)
+    .filter((id): id is string => typeof id === "string");
+
+  const idByExternalId = new Map<string, string>();
+  if (externalIds.length > 0) {
+    const rows = await prisma.workout.findMany({
+      where: { userId, externalId: { in: externalIds } },
+      select: { id: true, externalId: true },
+    });
+    for (const row of rows) {
+      if (row.externalId) idByExternalId.set(row.externalId, row.id);
+    }
+  }
+
+  for (const p of inserted) {
+    const startedAt =
+      p.row.startedAt instanceof Date
+        ? p.row.startedAt
+        : new Date(p.row.startedAt as string);
+    if (Number.isNaN(startedAt.getTime())) continue;
+    const externalId = p.dedupKey?.externalId;
+    await emitDataArrival({
+      userId,
+      kind: "workout",
+      newestSampleAt: startedAt,
+      insertedCount: 1,
+      refId: externalId ? idByExternalId.get(externalId) : undefined,
+      source: "batch",
+    });
+  }
+}
+
 export const POST = apiHandler(withIdempotency<[NextRequest]>(postBatch));
 
 async function postBatch(request: NextRequest): Promise<Response> {
@@ -673,6 +733,20 @@ async function postBatch(request: NextRequest): Promise<Response> {
   // workout-derived metrics.
   if (insertedCount > 0) {
     invalidateUserMeasurements(user.id);
+
+    // v1.31.0 — the workout arm of the data-arrival spine.
+    //
+    // Unlike every other kind, workouts emit ONE arrival PER workout, keyed by
+    // the workout's own id. Two sessions in a day are two events and each
+    // deserves its own reaction; a day-scoped key would silently collapse the
+    // second one.
+    //
+    // That needs the ids, and `createMany` does not return them, so this costs
+    // one extra indexed lookup — but only on a batch that actually inserted
+    // something, and only for entries carrying an external id. The classifier
+    // still runs per emit, so a historical import walks straight into
+    // `backfill` and enqueues nothing regardless of how many rows it wrote.
+    void emitWorkoutArrivals(user.id, prepared, results).catch(() => {});
   }
 
   return apiSuccess({