feat(apple-health): show when each type last arrived and how

The Apple Health card listed each type's newest sample. A type the phone
only hands over on "Sync all" can carry a reading from this morning and
never have arrived in the background, so the sample time could not answer
whether a type flows on its own.

The batch route's per-type ledger now records, in one upsert per batch,
when each Apple Health type last arrived, the trigger that carried it
(including the new `manual`), and when a batch last brought a new or
updated sample of it. Duplicate-only re-sends move the arrival but keep the
last new value; scoped credentials and MANUAL rows write nothing, the same
rule as the sync checkpoint. A ledger failure leaves a warn line and never
fails the ingest.

`GET /api/integrations/healthkit` lays `lastReceivedAt`, `lastTrigger` and
`lastNewSampleAt` over each `metricFreshness` entry (null for a type no live
sync has carried, absent on the workouts entry and when the ledger read
failed), and the card shows them as a second line per type.

The per-type newest-sample read walks live types through
`listLiveMeasurementTypes` and probes each type's newest row instead of
grouping every live row of the source.

src/app/api/integrations/healthkit/__tests__/route.test.ts

@@ -8,12 +8,20 @@ vi.mock("@/lib/db", () => ({
     auditLog: { create: vi.fn() },
     // Per-metric freshness + first-run sync progress for the APPLE_HEALTH
     // source (#778).
-    measurement: { groupBy: vi.fn(), aggregate: vi.fn() },
+    measurement: { aggregate: vi.fn() },
     workout: { groupBy: vi.fn(), aggregate: vi.fn() },
+    // v1.42 — the newest sample per live type is one raw probe per type, and
+    // the per-type arrival ledger (#1173) is read beside it.
+    $queryRaw: vi.fn(),
+    healthKitTypeSync: { findMany: vi.fn() },
   },
   toJson: <T>(v: T) => v,
 }));
 
+vi.mock("@/lib/measurements/live-types", () => ({
+  listLiveMeasurementTypes: vi.fn(),
+}));
+
 vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
 
 vi.mock("@/lib/auth/audit", () => ({
@@ -38,6 +46,7 @@ vi.mock("next/headers", () => ({
 import { GET, PATCH } from "../route";
 import { prisma } from "@/lib/db";
 import { getSession } from "@/lib/auth/session";
+import { listLiveMeasurementTypes } from "@/lib/measurements/live-types";
 import { APPLE_HEALTH_DATA_STALE_AFTER_MS } from "@/lib/integrations/sync-verdict";
 
 const SESSION_OK = {
@@ -53,7 +62,9 @@ beforeEach(() => {
   } as never);
   vi.mocked(prisma.user.update).mockResolvedValue({} as never);
   vi.mocked(prisma.auditLog.create).mockResolvedValue({} as never);
-  vi.mocked(prisma.measurement.groupBy).mockResolvedValue([] as never);
+  vi.mocked(listLiveMeasurementTypes).mockResolvedValue([]);
+  vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never);
+  vi.mocked(prisma.healthKitTypeSync.findMany).mockResolvedValue([] as never);
   vi.mocked(prisma.workout.groupBy).mockResolvedValue([] as never);
   vi.mocked(prisma.measurement.aggregate).mockResolvedValue({
     _count: { _all: 0 },
@@ -281,26 +292,20 @@ describe("GET /api/integrations/healthkit — sync health", () => {
       healthKitConfigJson: null,
       healthKitLastSyncedAt: ago(60 * 60 * 1000),
     } as never);
-    vi.mocked(prisma.measurement.groupBy).mockResolvedValue([
-      {
-        source: "APPLE_HEALTH",
-        type: "RESPIRATORY_RATE",
-        _max: { measuredAt: ago(30 * DAY) },
-      },
-      {
-        source: "APPLE_HEALTH",
-        type: "PULSE",
-        _max: { measuredAt: ago(60 * 60 * 1000) },
-      },
+    vi.mocked(listLiveMeasurementTypes).mockResolvedValue([
+      "RESPIRATORY_RATE",
+      "PULSE",
+    ]);
+    vi.mocked(prisma.$queryRaw).mockResolvedValue([
+      { type: "RESPIRATORY_RATE", last_seen: ago(30 * DAY) },
+      { type: "PULSE", last_seen: ago(60 * 60 * 1000) },
     ] as never);
 
     const body = await read();
     // Only the Apple Health source is read.
-    const groupByArgs = vi.mocked(prisma.measurement.groupBy).mock
-      .calls[0]?.[0] as unknown as {
-      where: { source: { in: string[] } };
-    };
-    expect(groupByArgs.where.source.in).toEqual(["APPLE_HEALTH"]);
+    expect(vi.mocked(listLiveMeasurementTypes).mock.calls[0]?.[1]).toEqual({
+      source: "APPLE_HEALTH",
+    });
     const byType = Object.fromEntries(
       body.metricFreshness.map((entry) => [entry.type, entry.stale]),
     );
@@ -382,11 +387,75 @@ describe("GET /api/integrations/healthkit — sync health", () => {
       healthKitConfigJson: null,
       healthKitLastSyncedAt: ago(DAY),
     } as never);
-    vi.mocked(prisma.measurement.groupBy).mockRejectedValue(
-      new Error("groupBy hiccup"),
+    vi.mocked(listLiveMeasurementTypes).mockRejectedValue(
+      new Error("live-type hiccup"),
     );
     const body = await read();
     expect(body.metricFreshness).toEqual([]);
     expect(body.syncHealth.verdict).toBe("fresh");
   });
+
+  // #1173 — when a type last ARRIVED and under which trigger, beside when its
+  // newest sample was taken. A type that only ever comes with "Sync all" is
+  // visible as `manual`; a type no live sync carried yet reads null.
+  it("lays the per-type arrival facts over the freshness entries", async () => {
+    vi.mocked(prisma.user.findUnique).mockResolvedValue({
+      healthKitConfigJson: null,
+      healthKitLastSyncedAt: ago(60 * 60 * 1000),
+    } as never);
+    vi.mocked(listLiveMeasurementTypes).mockResolvedValue([
+      "RESPIRATORY_RATE",
+      "PULSE",
+    ]);
+    vi.mocked(prisma.$queryRaw).mockResolvedValue([
+      { type: "RESPIRATORY_RATE", last_seen: ago(2 * DAY) },
+      { type: "PULSE", last_seen: ago(60 * 60 * 1000) },
+    ] as never);
+    vi.mocked(prisma.workout.groupBy).mockResolvedValue([
+      { source: "APPLE_HEALTH", _max: { startedAt: ago(DAY) } },
+    ] as never);
+    const receivedAt = ago(3 * 60 * 60 * 1000);
+    vi.mocked(prisma.healthKitTypeSync.findMany).mockResolvedValue([
+      {
+        type: "RESPIRATORY_RATE",
+        lastReceivedAt: receivedAt,
+        lastTrigger: "manual",
+        lastNewSampleAt: null,
+      },
+    ] as never);
+
+    const body = await read();
+    const byType = Object.fromEntries(
+      body.metricFreshness.map((entry) => [entry.type, entry]),
+    );
+    expect(byType.RESPIRATORY_RATE).toMatchObject({
+      lastReceivedAt: receivedAt.toISOString(),
+      lastTrigger: "manual",
+      lastNewSampleAt: null,
+    });
+    expect(byType.PULSE).toMatchObject({
+      lastReceivedAt: null,
+      lastTrigger: null,
+      lastNewSampleAt: null,
+    });
+    // Workouts ride their own batch route, which the ledger does not cover.
+    expect(byType.WORKOUTS).not.toHaveProperty("lastReceivedAt");
+  });
+
+  it("leaves the arrival facts off when the ledger read fails", async () => {
+    vi.mocked(prisma.user.findUnique).mockResolvedValue({
+      healthKitConfigJson: null,
+      healthKitLastSyncedAt: ago(DAY),
+    } as never);
+    vi.mocked(listLiveMeasurementTypes).mockResolvedValue(["PULSE"]);
+    vi.mocked(prisma.$queryRaw).mockResolvedValue([
+      { type: "PULSE", last_seen: ago(DAY) },
+    ] as never);
+    vi.mocked(prisma.healthKitTypeSync.findMany).mockRejectedValue(
+      new Error("ledger hiccup"),
+    );
+    const body = await read();
+    expect(body.metricFreshness).toHaveLength(1);
+    expect(body.metricFreshness[0]).not.toHaveProperty("lastReceivedAt");
+  });
 });

src/app/api/integrations/healthkit/route.ts

@@ -21,6 +21,10 @@ import { annotate } from "@/lib/logging/context";
 import { prisma, toJson } from "@/lib/db";
 import { auditLog } from "@/lib/auth/audit";
 import { getAppleHealthSyncProgress } from "@/lib/integrations/apple-health-progress";
+import {
+  getHealthKitTypeArrivals,
+  withHealthKitArrivals,
+} from "@/lib/integrations/healthkit-type-sync";
 import { getSourceFreshness } from "@/lib/integrations/metric-freshness";
 import {
   classifyMetricFreshness,
@@ -166,9 +170,13 @@ export const GET = apiHandler(async () => {
     legacyLastSyncedAt: lastSyncedAt,
     cadence: PUSH_CADENCE,
   });
-  const samples = await getSourceFreshness(user.id, "APPLE_HEALTH").catch(
-    () => [],
-  );
+  // #1173 — when each type last ARRIVED and under which trigger, beside when
+  // its newest sample was taken. Fail-soft to null, which leaves the entries
+  // without arrival fields rather than reporting every type as never received.
+  const [samples, arrivals] = await Promise.all([
+    getSourceFreshness(user.id, "APPLE_HEALTH").catch(() => []),
+    getHealthKitTypeArrivals(user.id).catch(() => null),
+  ]);
   // Issue #778 — a first-run backfill used to be invisible from the web: the
   // card said "last sync 2 minutes ago" while giving no sense of how much had
   // arrived or how far back it reached. These two figures (row count + oldest
@@ -186,7 +194,10 @@ export const GET = apiHandler(async () => {
     lastSyncTrigger,
     lastBackgroundSyncAt,
     syncHealth,
-    metricFreshness: classifyMetricFreshness(samples, syncHealth.verdict),
+    metricFreshness: withHealthKitArrivals(
+      classifyMetricFreshness(samples, syncHealth.verdict),
+      arrivals,
+    ),
     syncProgress,
   });
 });

src/app/api/integrations/status/__tests__/route.test.ts

@@ -13,11 +13,16 @@ vi.mock("@/lib/db", () => ({
     fitbitConnection: { findUnique: vi.fn(async () => null) },
     googleHealthConnection: { findUnique: vi.fn(async () => null) },
     moodEntry: { count: vi.fn(async () => 0) },
-    measurement: { groupBy: vi.fn(async () => []) },
+    $queryRaw: vi.fn(async () => []),
     workout: { groupBy: vi.fn(async () => []) },
   },
 }));
 
+// v1.42 — the per-source type walk that replaced the measurement groupBy.
+vi.mock("@/lib/measurements/live-types", () => ({
+  listLiveMeasurementTypes: vi.fn(async () => []),
+}));
+
 vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
 
 vi.mock("@/lib/api-response", () => ({
@@ -53,6 +58,7 @@ vi.mock("@/lib/oura/credentials", () => ({
 
 import { GET } from "../route";
 import { prisma } from "@/lib/db";
+import { listLiveMeasurementTypes } from "@/lib/measurements/live-types";
 import { ATTEMPT_STALE_AFTER_MS } from "@/lib/integrations/sync-verdict";
 
 const userFind = prisma.user.findUnique as ReturnType<typeof vi.fn>;
@@ -315,7 +321,7 @@ describe("/api/integrations/status — the freshness read failed", () => {
   });
 
   it("reports the degradation instead of passing an empty list off as an answer", async () => {
-    (prisma.measurement.groupBy as ReturnType<typeof vi.fn>).mockRejectedValue(
+    vi.mocked(listLiveMeasurementTypes).mockRejectedValue(
       new Error("statement timeout"),
     );
 
@@ -331,9 +337,7 @@ describe("/api/integrations/status — the freshness read failed", () => {
   });
 
   it("reports no degradation when the read simply found nothing", async () => {
-    (prisma.measurement.groupBy as ReturnType<typeof vi.fn>).mockResolvedValue(
-      [],
-    );
+    vi.mocked(listLiveMeasurementTypes).mockResolvedValue([]);
     (prisma.workout.groupBy as ReturnType<typeof vi.fn>).mockResolvedValue([]);
 
     const envelope = await fetchEnvelope();