fix(release): harden v1.31 arrival contracts

src/app/api/import/__tests__/route.test.ts

@@ -10,7 +10,13 @@ vi.mock("@/lib/api-handler", () => ({
 
 vi.mock("@/lib/db", () => ({
   prisma: {
-    measurement: { create: vi.fn(), upsert: vi.fn() },
+    measurement: {
+      create: vi.fn(),
+      createManyAndReturn: vi.fn(),
+      update: vi.fn(),
+      findUnique: vi.fn(),
+      upsert: vi.fn(),
+    },
     moodEntry: { create: vi.fn(), upsert: vi.fn() },
   },
 }));
@@ -27,6 +33,13 @@ vi.mock("@/lib/logging/context", () => ({
   annotate: vi.fn(),
 }));
 
+vi.mock("@/lib/arrivals/measurement-emit", () => ({
+  emitInsertedMeasurementArrivals: vi.fn(),
+}));
+
+vi.mock("@/lib/daily/morning-refresh-trigger", () => ({
+  maybeEnqueueMorningRefresh: vi.fn(),
+}));
 // v1.4.39.1 — surface the rollup hook so the new regression test can
 // assert that the import path now folds the persistent rollup tier
 // for each touched (type, day). Mood rollup mock stays no-op because
@@ -50,9 +63,14 @@ import { POST } from "../route";
 import { checkRateLimit } from "@/lib/rate-limit";
 import { prisma } from "@/lib/db";
 import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups";
+import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
+import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
 
 beforeEach(() => {
   vi.resetAllMocks();
+  vi.mocked(prisma.measurement.findUnique).mockResolvedValue(null);
+  vi.mocked(emitInsertedMeasurementArrivals).mockResolvedValue(undefined);
+  vi.mocked(maybeEnqueueMorningRefresh).mockResolvedValue(undefined);
 });
 
 interface ApiErrorEnvelope {
@@ -242,6 +260,173 @@ describe("POST /api/import — measurement rollup hook (v1.4.39.1)", () => {
   });
 });
 
+describe("POST /api/import — arrival wiring", () => {
+  beforeEach(() => {
+    vi.mocked(checkRateLimit).mockResolvedValue({
+      allowed: true,
+      remaining: 4,
+      resetAt: new Date(Date.now() + 1000),
+    } as never);
+  });
+
+  it("emits and refreshes only a newly inserted sleep row", async () => {
+    const measuredAt = new Date("2026-05-14T06:00:00.000Z");
+    const created = {
+      id: "sleep-new",
+      type: "SLEEP_DURATION" as const,
+      measuredAt,
+    };
+    vi.mocked(prisma.measurement.create).mockResolvedValue(created as never);
+
+    const response = await POST(
+      new NextRequest("http://localhost/api/import", {
+        method: "POST",
+        body: JSON.stringify({
+          measurements: [
+            {
+              type: "SLEEP_DURATION",
+              value: 480,
+              unit: "min",
+              measuredAt: measuredAt.toISOString(),
+            },
+          ],
+        }),
+        headers: { "content-type": "application/json" },
+      }),
+    );
+
+    expect(response.status).toBeLessThan(400);
+    expect(emitInsertedMeasurementArrivals).toHaveBeenCalledWith(
+      "u-1",
+      [created],
+      "json_import",
+    );
+    expect(maybeEnqueueMorningRefresh).toHaveBeenCalledWith("u-1", [
+      measuredAt,
+    ]);
+  });
+
+  it("emits and refreshes an external-id row only when INSERT returns it", async () => {
+    const measuredAt = new Date("2026-05-14T06:00:00.000Z");
+    const inserted = {
+      id: "sleep-inserted",
+      type: "SLEEP_DURATION" as const,
+      measuredAt,
+    };
+    vi.mocked(prisma.measurement.createManyAndReturn).mockResolvedValue([
+      inserted,
+    ] as never);
+
+    const response = await POST(
+      new NextRequest("http://localhost/api/import", {
+        method: "POST",
+        body: JSON.stringify({
+          measurements: [
+            {
+              type: "SLEEP_DURATION",
+              value: 480,
+              unit: "min",
+              measuredAt: measuredAt.toISOString(),
+              externalId: "sleep-new",
+            },
+          ],
+        }),
+        headers: { "content-type": "application/json" },
+      }),
+    );
+
+    expect(response.status).toBeLessThan(400);
+    expect(emitInsertedMeasurementArrivals).toHaveBeenCalledWith(
+      "u-1",
+      [inserted],
+      "json_import",
+    );
+    expect(maybeEnqueueMorningRefresh).toHaveBeenCalledWith("u-1", [
+      measuredAt,
+    ]);
+  });
+
+  it("does not emit or refresh when INSERT loses a race and returns no row", async () => {
+    vi.mocked(prisma.measurement.findUnique).mockResolvedValue(null);
+    vi.mocked(prisma.measurement.createManyAndReturn).mockResolvedValue(
+      [] as never,
+    );
+    vi.mocked(prisma.measurement.update).mockResolvedValue({
+      id: "raced-winner",
+      type: "SLEEP_DURATION",
+      measuredAt: new Date("2026-05-14T06:00:00.000Z"),
+    } as never);
+    vi.mocked(prisma.measurement.upsert).mockResolvedValue({
+      id: "raced-winner",
+      type: "SLEEP_DURATION",
+      measuredAt: new Date("2026-05-14T06:00:00.000Z"),
+    } as never);
+
+    const response = await POST(
+      new NextRequest("http://localhost/api/import", {
+        method: "POST",
+        body: JSON.stringify({
+          measurements: [
+            {
+              type: "SLEEP_DURATION",
+              value: 480,
+              unit: "min",
+              measuredAt: "2026-05-14T06:00:00.000Z",
+              externalId: "sleep-raced",
+            },
+          ],
+        }),
+        headers: { "content-type": "application/json" },
+      }),
+    );
+
+    expect(response.status).toBeLessThan(400);
+    expect(emitInsertedMeasurementArrivals).toHaveBeenCalledWith(
+      "u-1",
+      [],
+      "json_import",
+    );
+    expect(maybeEnqueueMorningRefresh).toHaveBeenCalledWith("u-1", []);
+  });
+
+  it("does not emit or refresh an existing external-id update", async () => {
+    vi.mocked(prisma.measurement.findUnique).mockResolvedValue({
+      id: "existing",
+    } as never);
+    vi.mocked(prisma.measurement.upsert).mockResolvedValue({
+      id: "existing",
+      type: "SLEEP_DURATION",
+      measuredAt: new Date("2026-05-14T06:00:00.000Z"),
+    } as never);
+
+    const response = await POST(
+      new NextRequest("http://localhost/api/import", {
+        method: "POST",
+        body: JSON.stringify({
+          measurements: [
+            {
+              type: "SLEEP_DURATION",
+              value: 480,
+              unit: "min",
+              measuredAt: "2026-05-14T06:00:00.000Z",
+              externalId: "sleep-existing",
+            },
+          ],
+        }),
+        headers: { "content-type": "application/json" },
+      }),
+    );
+
+    expect(response.status).toBeLessThan(400);
+    expect(emitInsertedMeasurementArrivals).toHaveBeenCalledWith(
+      "u-1",
+      [],
+      "json_import",
+    );
+    expect(maybeEnqueueMorningRefresh).toHaveBeenCalledWith("u-1", []);
+  });
+});
+
 // v1.17.1 — two real bugs the data-portability audit found in the JSON
 // import. Regression guards so a future refactor can't silently re-open
 // them.
@@ -323,8 +508,11 @@ describe("POST /api/import — measurement externalId dedup (v1.17.1)", () => {
     } as never);
   });
 
-  it("upserts (not creates) when a measurement carries an externalId", async () => {
-    vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never);
+  it("updates in place when an external-id INSERT returns no row", async () => {
+    vi.mocked(prisma.measurement.createManyAndReturn).mockResolvedValue(
+      [] as never,
+    );
+    vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
 
     const response = await POST(
       new NextRequest("http://localhost/api/import", {
@@ -345,9 +533,11 @@ describe("POST /api/import — measurement externalId dedup (v1.17.1)", () => {
     );
 
     expect(response.status).toBeLessThan(400);
-    expect(vi.mocked(prisma.measurement.upsert)).toHaveBeenCalledTimes(1);
+    expect(
+      vi.mocked(prisma.measurement.createManyAndReturn),
+    ).toHaveBeenCalledTimes(1);
     expect(vi.mocked(prisma.measurement.create)).not.toHaveBeenCalled();
-    const arg = vi.mocked(prisma.measurement.upsert).mock.calls[0][0];
+    const arg = vi.mocked(prisma.measurement.update).mock.calls[0][0];
     expect(arg.where).toEqual({
       userId_type_source_externalId: {
         userId: "u-1",

src/app/api/import/csv/__tests__/route.test.ts

@@ -11,14 +11,14 @@ vi.mock("@/lib/api-handler", () => ({
 vi.mock("@/lib/db", () => {
   const measurement = {
     findMany: vi.fn(),
-    createMany: vi.fn(),
+    createManyAndReturn: vi.fn(),
     updateMany: vi.fn(),
   };
   return {
     prisma: {
       measurement,
       // Run the batched write callback against the same measurement mock so
-      // createMany / updateMany calls are observable.
+      // createManyAndReturn / updateMany calls are observable.
       $transaction: vi.fn(async (fn: (tx: unknown) => unknown) =>
         fn({ measurement }),
       ),
@@ -99,14 +99,15 @@ beforeEach(() => {
   // db mock factory (resetAllMocks clears its implementation), so restore it.
   const measurement = prisma.measurement as unknown as {
     findMany: ReturnType<typeof vi.fn>;
-    createMany: ReturnType<typeof vi.fn>;
+    createManyAndReturn: ReturnType<typeof vi.fn>;
     updateMany: ReturnType<typeof vi.fn>;
   };
   measurement.findMany.mockResolvedValue([]);
-  // Echo the attempted chunk size, matching Postgres semantics when no row
-  // collides — the route now reconciles `inserted` against this count.
-  measurement.createMany.mockImplementation(
-    async (arg: { data: unknown[] }) => ({ count: arg.data.length }),
+  // Echo the attempted rows, matching Postgres semantics when no row
+  // collides — the route reconciles `inserted` against this result.
+  measurement.createManyAndReturn.mockImplementation(
+    async (arg: { data: Array<Record<string, unknown>> }) =>
+      arg.data.map((row, index) => ({ ...row, id: `inserted-${index}` })),
   );
   measurement.updateMany.mockResolvedValue({ count: 0 });
   (
@@ -139,7 +140,7 @@ describe("POST /api/import/csv — rate limit", () => {
 const mMeasurement = () =>
   prisma.measurement as unknown as {
     findMany: ReturnType<typeof vi.fn>;
-    createMany: ReturnType<typeof vi.fn>;
+    createManyAndReturn: ReturnType<typeof vi.fn>;
     updateMany: ReturnType<typeof vi.fn>;
   };
 
@@ -147,7 +148,7 @@ describe("POST /api/import/csv — fatal header error", () => {
   it("returns 422 when a required column is missing", async () => {
     const res = await POST(csvRequest("type,value,unit\nWEIGHT,80,kg"));
     expect(res.status).toBe(422);
-    expect(mMeasurement().createMany).not.toHaveBeenCalled();
+    expect(mMeasurement().createManyAndReturn).not.toHaveBeenCalled();
   });
 });
 
@@ -173,15 +174,15 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
     const reasons = body.data?.rows.map((r) => r.reason);
     expect(reasons).toContain("unknown_type");
     expect(reasons).toContain("missing_timezone_offset");
-    // One bulk createMany carrying the single valid survivor.
-    expect(mMeasurement().createMany).toHaveBeenCalledTimes(1);
-    const arg = mMeasurement().createMany.mock.calls[0][0];
+    // One bulk createManyAndReturn carrying the single valid survivor.
+    expect(mMeasurement().createManyAndReturn).toHaveBeenCalledTimes(1);
+    const arg = mMeasurement().createManyAndReturn.mock.calls[0][0];
     expect(arg.data).toHaveLength(1);
     // Rollup re-fold fired for the touched (type, day).
     expect(vi.mocked(recomputeBucketsForMeasurement)).toHaveBeenCalled();
   });
 
-  it("batches many rows into a single createMany (no per-row round-trip)", async () => {
+  it("batches many rows into a single createManyAndReturn", async () => {
     const rows = [HEADER];
     for (let i = 0; i < 50; i++) {
       const mm = String(i).padStart(2, "0"); // unique minute → unique measuredAt
@@ -191,9 +192,11 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
     expect(res.status).toBe(200);
     const body = (await res.json()) as CsvEnvelope;
     expect(body.data?.inserted).toBe(50);
-    // 50 rows, one createMany call — the whole point of the batch.
-    expect(mMeasurement().createMany).toHaveBeenCalledTimes(1);
-    expect(mMeasurement().createMany.mock.calls[0][0].data).toHaveLength(50);
+    // 50 rows, one bulk call — the whole point of the batch.
+    expect(mMeasurement().createManyAndReturn).toHaveBeenCalledTimes(1);
+    expect(
+      mMeasurement().createManyAndReturn.mock.calls[0][0].data,
+    ).toHaveLength(50);
   });
 
   it("dryRun previews without writing and reports projected inserts", async () => {
@@ -208,7 +211,7 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
     expect(body.data?.dryRun).toBe(true);
     expect(body.data?.inserted).toBe(1);
     expect(body.data?.rows[0].status).toBe("inserted");
-    expect(mMeasurement().createMany).not.toHaveBeenCalled();
+    expect(mMeasurement().createManyAndReturn).not.toHaveBeenCalled();
     expect(vi.mocked(recomputeBucketsForMeasurement)).not.toHaveBeenCalled();
   });
 
@@ -227,7 +230,7 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
     expect(body.data?.inserted).toBe(0);
     expect(body.data?.rows[0].status).toBe("updated");
     expect(mMeasurement().updateMany).toHaveBeenCalledTimes(1);
-    expect(mMeasurement().createMany).not.toHaveBeenCalled();
+    expect(mMeasurement().createManyAndReturn).not.toHaveBeenCalled();
   });
 
   it("resurrects a tombstoned externalId row on re-import (deletedAt: null in update)", async () => {
@@ -264,7 +267,7 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
     expect(body.data?.inserted).toBe(1);
     expect(body.data?.updated).toBe(0);
     expect(body.data?.rows[0].status).toBe("inserted");
-    expect(mMeasurement().createMany).toHaveBeenCalledTimes(1);
+    expect(mMeasurement().createManyAndReturn).toHaveBeenCalledTimes(1);
   });
 
   it("counts a pre-existing natural-key row as skipped/duplicate", async () => {
@@ -282,7 +285,7 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
       status: "skipped",
       reason: "duplicate",
     });
-    expect(mMeasurement().createMany).not.toHaveBeenCalled();
+    expect(mMeasurement().createManyAndReturn).not.toHaveBeenCalled();
   });
 
   it("collapses an in-file duplicate (same type+measuredAt) to one insert + one duplicate", async () => {
@@ -299,16 +302,21 @@ describe("POST /api/import/csv — batched write + per-row envelope", () => {
     expect(body.data?.inserted).toBe(1);
     expect(body.data?.skipped).toBe(1);
     // Only the first survivor reaches the bulk insert.
-    expect(mMeasurement().createMany.mock.calls[0][0].data).toHaveLength(1);
+    expect(
+      mMeasurement().createManyAndReturn.mock.calls[0][0].data,
+    ).toHaveLength(1);
   });
 
-  it("reconciles `inserted` against the createMany count when skipDuplicates absorbs a race", async () => {
+  it("reconciles `inserted` against returned rows after a race", async () => {
     // Two fresh rows attempted, but a concurrent double-submit already
     // landed one of them — `skipDuplicates` absorbs the conflict and the
     // count comes back short. The envelope must sum to the DB truth: one
     // inserted, one downgraded to skipped/duplicate.
-    mMeasurement().createMany.mockImplementation(
-      async (arg: { data: unknown[] }) => ({ count: arg.data.length - 1 }),
+    mMeasurement().createManyAndReturn.mockImplementation(
+      async (arg: { data: Array<Record<string, unknown>> }) =>
+        arg.data
+          .slice(0, -1)
+          .map((row, index) => ({ ...row, id: `inserted-${index}` })),
     );
 
     const res = await POST(

src/app/api/import/csv/route.ts

@@ -21,6 +21,8 @@ import type {
 } from "@/generated/prisma/client";
 import { Prisma } from "@/generated/prisma/client";
 import type { NormalisedMeasurementRow } from "@/lib/import/csv-measurements";
+import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
+import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
 
 /**
  * CSV measurement import (v1.17.1).
@@ -231,22 +233,29 @@ export const POST = apiHandler(async (request: NextRequest) => {
       }
     }
 
-    // Bulk write: createMany the survivors (chunked under the PG parameter
-    // cap, skipDuplicates to absorb a concurrent double-submit race), then
-    // grouped updateMany for the pre-existing externalId rows.
+    // Bulk write: createManyAndReturn the survivors (chunked under the PG
+    // parameter cap, skipDuplicates to absorb a concurrent double-submit
+    // race), then grouped updateMany for pre-existing externalId rows.
     const toCreate: Prisma.MeasurementCreateManyInput[] = [
       ...[...extInserts.values()].map(buildCreateData),
       ...plainInserts.map(buildCreateData),
     ];
     const CREATE_CHUNK = 200;
     let createdCount = 0;
+    const createdRows: Array<{
+      id: string;
+      type: MeasurementType;
+      measuredAt: Date;
+    }> = [];
     await prisma.$transaction(async (tx) => {
       for (let i = 0; i < toCreate.length; i += CREATE_CHUNK) {
-        const res = await tx.measurement.createMany({
+        const rows = await tx.measurement.createManyAndReturn({
           data: toCreate.slice(i, i + CREATE_CHUNK),
           skipDuplicates: true,
+          select: { id: true, type: true, measuredAt: true },
         });
-        createdCount += res.count;
+        createdRows.push(...rows);
+        createdCount += rows.length;
       }
       for (const m of extUpdates.values()) {
         await tx.measurement.updateMany({
@@ -274,6 +283,18 @@ export const POST = apiHandler(async (request: NextRequest) => {
       }
     });
 
+    void emitInsertedMeasurementArrivals(
+      userId,
+      createdRows,
+      "csv_import",
+    ).catch(() => {});
+    void maybeEnqueueMorningRefresh(
+      userId,
+      createdRows
+        .filter((row) => row.type === "SLEEP_DURATION")
+        .map((row) => row.measuredAt),
+    ).catch(() => {});
+
     // Reconcile `inserted` against what `createMany` actually wrote
     // (mirrors the batch route's raced-duplicate downgrade). Under
     // `skipDuplicates` a conflicting row is silently absorbed — the key is

src/app/api/import/route.ts

@@ -25,6 +25,8 @@ import {
   recomputeBucketsForMeasurement,
 } from "@/lib/rollups/measurement-rollups";
 import type { MeasurementType } from "@/generated/prisma/client";
+import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
+import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
 
 // Derived from canonical enum so round-trip export → import covers every
 // type. Previous hardcoded subset silently dropped 4 of 11 types
@@ -122,48 +124,67 @@ export const POST = apiHandler(async (request: NextRequest) => {
     type: MeasurementType;
     measuredAt: Date;
   }> = [];
+  const insertedMeasurements: Array<{
+    id: string;
+    type: MeasurementType;
+    measuredAt: Date;
+  }> = [];
   if (data.measurements?.length) {
     for (const m of data.measurements) {
       try {
         // `measuredAt` is already a `Date` (validateEntryInstant transform).
         const measuredAt = m.measuredAt;
         if (m.externalId) {
-          // v1.17.1 — idempotent re-import keyed on the source-stable id, so
-          // re-uploading the same export updates in place rather than minting
-          // a duplicate. Mirrors the mood path. The unique key already exists
-          // (`userId_type_source_externalId`, schema.prisma:942).
-          await prisma.measurement.upsert({
-            where: {
-              userId_type_source_externalId: {
+          const notesEncrypted = encryptNote(m.notes || null);
+          // INSERT ... RETURNING is the only insertion proof. A concurrent
+          // importer can win this key after any existence probe, so attempt
+          // the insert first and update only when the statement returns no row.
+          const created = await prisma.measurement.createManyAndReturn({
+            data: [
+              {
                 userId,
                 type: m.type,
+                value: m.value,
+                unit: m.unit,
                 source: "IMPORT",
                 externalId: m.externalId,
+                measuredAt,
+                notes: null,
+                notesEncrypted,
+                glucoseContext: m.glucoseContext ?? null,
               },
-            },
-            update: {
-              value: m.value,
-              unit: m.unit,
-              measuredAt,
-              notes: null,
-              notesEncrypted: encryptNote(m.notes || null),
-              glucoseContext: m.glucoseContext ?? null,
-            },
-            create: {
-              userId,
-              type: m.type,
-              value: m.value,
-              unit: m.unit,
-              source: "IMPORT",
-              externalId: m.externalId,
-              measuredAt,
-              notes: null,
-              notesEncrypted: encryptNote(m.notes || null),
-              glucoseContext: m.glucoseContext ?? null,
-            },
+            ],
+            skipDuplicates: true,
+            select: { id: true, type: true, measuredAt: true },
           });
+          const inserted = created[0];
+          if (inserted) {
+            insertedMeasurements.push(inserted);
+          } else {
+            // A pre-existing row or a raced insert owns the key. Preserve
+            // idempotent re-import semantics without classifying this update
+            // as an arrival.
+            await prisma.measurement.update({
+              where: {
+                userId_type_source_externalId: {
+                  userId,
+                  type: m.type,
+                  source: "IMPORT",
+                  externalId: m.externalId,
+                },
+              },
+              data: {
+                value: m.value,
+                unit: m.unit,
+                measuredAt,
+                notes: null,
+                notesEncrypted,
+                glucoseContext: m.glucoseContext ?? null,
+              },
+            });
+          }
         } else {
-          await prisma.measurement.create({
+          const inserted = await prisma.measurement.create({
             data: {
               userId,
               type: m.type,
@@ -175,7 +196,9 @@ export const POST = apiHandler(async (request: NextRequest) => {
               notesEncrypted: encryptNote(m.notes || null),
               glucoseContext: m.glucoseContext ?? null,
             },
+            select: { id: true, type: true, measuredAt: true },
           });
+          insertedMeasurements.push(inserted);
         }
         touchedMeasurements.push({
           type: m.type as MeasurementType,
@@ -189,6 +212,18 @@ export const POST = apiHandler(async (request: NextRequest) => {
     }
   }
 
+  void emitInsertedMeasurementArrivals(
+    userId,
+    insertedMeasurements,
+    "json_import",
+  ).catch(() => {});
+  void maybeEnqueueMorningRefresh(
+    userId,
+    insertedMeasurements
+      .filter((row) => row.type === "SLEEP_DURATION")
+      .map((row) => row.measuredAt),
+  ).catch(() => {});
+
   // Import mood entries
   if (data.moodEntries?.length) {
     for (const e of data.moodEntries) {