feat(arrivals): add the data-arrival event spine

Every ingest path in the product can now emit a typed, salience-classified
event into pg-boss, consumed by one worker that claims a durable per-day
reaction marker and fans out to the reaction surfaces.

The recency classifier runs at the seam, before anything is enqueued, so a
mass backfill — a multi-year Apple Health export, a provider re-sync, a
30-day catch-up — emits nothing at all. Non-salient writes are annotated and
dropped.

The worker makes no provider call by construction: it imports no provider
module, writes only markers, and enqueues nothing that is not budget-gated
downstream. Business refusals return a status rather than throwing, so a
refusal is never retried against a ceiling that will not move until the local
day rolls over.

Seams: sleep (one hook, inherited by all seven transports), workouts (the
batch route, the four provider syncs, the export importer), weight and blood
pressure (both measurement routes), labs (manual, OCR commit, document fact
commit).

The provider workout syncs could not previously tell a created row from an
updated one, so a re-sync of a recent session would have re-fired on every
poll; they now recover the distinction from the row the upsert already
returns.

Reaction lines are AES-256-GCM at rest, following the InsightNarrative
precedent, and the column is registered for key rotation. Markers are pruned
after 14 days by a daily maintenance job.

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/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({