feat(scripts): drain per-sample cumulative HealthKit rows into daily-aggregated rows

Idempotent maintenance tool that collapses pre-Option-A per-sample
APPLE_HEALTH cumulative rows (steps, active energy, flights climbed,
walking + running distance, time in daylight) into one row per day
per type, keyed by the locked dailyStatsExternalId shape. Per
user × type × calendar-day (anchored to User.timezone): SUM the values,
UPSERT a daily row at canonical midday-local (matching the Withings
activity-sync convention), DELETE the per-sample rows.

Three entry points share src/lib/measurements/drain-per-sample-cumulative.ts:
  - Library helper for unit + integration tests
  - CLI at scripts/drain-per-sample-cumulative.ts with --user / --all
    + --confirm flags (default dry-run)
  - Admin endpoint POST /api/admin/drain-per-sample-cumulative gated
    by requireAdmin() with dryRun defaulting to true

Operator runs the drain once after the iOS TestFlight build adopts
HealthKitStatisticsService.swift. Re-running on a fully-collapsed
account reports zero buckets touched.

Tests cover bucketing semantics, the canonical-timestamp math across
DST + half-hour zones, idempotent re-runs, dry-run preview, spot-metric
exclusion, and already-collapsed-row skip via real-Postgres.

scripts/drain-per-sample-cumulative.ts

@@ -0,0 +1,117 @@
+#!/usr/bin/env tsx
+/**
+ * scripts/drain-per-sample-cumulative.ts
+ *
+ * One-shot maintenance tool that collapses pre-Option-A per-sample
+ * APPLE_HEALTH cumulative rows into one row per day per cumulative
+ * type. See R-A §6 / v1.4.30 §"Cutover sequence".
+ *
+ * Scope per `CUMULATIVE_HK_TYPES`:
+ *   ACTIVITY_STEPS, ACTIVE_ENERGY_BURNED, FLIGHTS_CLIMBED,
+ *   WALKING_RUNNING_DISTANCE, TIME_IN_DAYLIGHT.
+ *
+ * Idempotent — re-running after a successful drain is a no-op.
+ *
+ * Usage:
+ *   # Dry-run for one user (default; no DB writes):
+ *   pnpm tsx scripts/drain-per-sample-cumulative.ts --user clx123
+ *
+ *   # Commit the drain for one user:
+ *   pnpm tsx scripts/drain-per-sample-cumulative.ts --user clx123 --confirm
+ *
+ *   # Dry-run for every user:
+ *   pnpm tsx scripts/drain-per-sample-cumulative.ts --all
+ *
+ *   # Commit the drain for every user (requires --confirm):
+ *   pnpm tsx scripts/drain-per-sample-cumulative.ts --all --confirm
+ *
+ * Operator runs this once after the iOS TestFlight build adopts
+ * `HealthKitStatisticsService.swift`. The drain runs idempotently; a
+ * re-run reports zero buckets collapsed.
+ */
+import "dotenv/config";
+
+import { prisma } from "@/lib/db";
+import { drainPerSampleCumulative } from "@/lib/measurements/drain-per-sample-cumulative";
+
+interface ParsedArgs {
+  userId?: string;
+  all: boolean;
+  confirm: boolean;
+}
+
+function parseArgs(argv: readonly string[]): ParsedArgs {
+  const out: ParsedArgs = { all: false, confirm: false };
+  for (let i = 0; i < argv.length; i++) {
+    const arg = argv[i];
+    if (arg === "--user") {
+      out.userId = argv[++i];
+    } else if (arg === "--all") {
+      out.all = true;
+    } else if (arg === "--confirm") {
+      out.confirm = true;
+    } else if (arg === "--help" || arg === "-h") {
+      console.log(
+        "Usage: pnpm tsx scripts/drain-per-sample-cumulative.ts [--user <id> | --all] [--confirm]",
+      );
+      process.exit(0);
+    } else {
+      console.error(`Unknown argument: ${arg}`);
+      process.exit(2);
+    }
+  }
+  return out;
+}
+
+async function main(): Promise<void> {
+  const args = parseArgs(process.argv.slice(2));
+
+  if (!args.userId && !args.all) {
+    console.error(
+      "Pick a scope: --user <id> (single user) or --all (every user). Default mode is dry-run; pass --confirm to commit.",
+    );
+    process.exit(2);
+  }
+  if (args.userId && args.all) {
+    console.error("--user and --all are mutually exclusive.");
+    process.exit(2);
+  }
+
+  const dryRun = !args.confirm;
+  if (dryRun) {
+    console.log(
+      "[drain] DRY RUN — no rows will be written. Pass --confirm to commit.",
+    );
+  } else {
+    console.log("[drain] COMMIT MODE — collapsing rows in place.");
+  }
+
+  try {
+    const summary = await drainPerSampleCumulative(prisma, {
+      userId: args.userId,
+      dryRun,
+    });
+
+    console.log("[drain] summary:");
+    console.log(JSON.stringify(summary.totals, null, 2));
+    if (summary.buckets.length > 0) {
+      console.log(
+        `[drain] first ${Math.min(5, summary.buckets.length)} buckets (of ${summary.buckets.length}):`,
+      );
+      for (const b of summary.buckets.slice(0, 5)) {
+        console.log(
+          `  - user=${b.userId} type=${b.type} day=${b.dateKey} samples=${b.perSampleCount} sum=${b.sumValue} extId=${b.externalId}`,
+        );
+      }
+    } else {
+      console.log("[drain] no buckets needed collapsing — already at parity.");
+    }
+  } finally {
+    await prisma.$disconnect();
+  }
+}
+
+main().catch((err: unknown) => {
+  console.error("[drain] failed:", err);
+  process.exit(1);
+});

src/app/api/admin/drain-per-sample-cumulative/route.ts

@@ -0,0 +1,93 @@
+/**
+ * `POST /api/admin/drain-per-sample-cumulative` — operator-triggered
+ * variant of `scripts/drain-per-sample-cumulative.ts`. Collapses
+ * pre-Option-A per-sample APPLE_HEALTH cumulative rows into one row
+ * per day per cumulative type. See `06-ios-responsibilities.md`
+ * §"Cumulative metrics: daily aggregation on iOS".
+ *
+ * Body:
+ *   { userId?: string, dryRun?: boolean }
+ *
+ *   - `userId` — single user mode. Omit to drain every user.
+ *   - `dryRun` — preview-only; no DB writes. Default: `true`. Set to
+ *     `false` to commit.
+ *
+ * Response shape (always 200 on success):
+ *   { processed, inserted (always 0), totals: { ... }, buckets: [...] }
+ *
+ * Admin gate via `requireAdmin()` (cookie-only — Bearer never elevates
+ * to admin per the v1.4.25 security boundary).
+ */
+import type { NextRequest } from "next/server";
+
+import { prisma } from "@/lib/db";
+import { apiHandler, requireAdmin } from "@/lib/api-handler";
+import { auditLog } from "@/lib/auth/audit";
+import {
+  apiError,
+  apiSuccess,
+  getClientIp,
+  safeJson,
+} from "@/lib/api-response";
+import { annotate } from "@/lib/logging/context";
+import { drainPerSampleCumulative } from "@/lib/measurements/drain-per-sample-cumulative";
+
+interface DrainBody {
+  userId?: string;
+  dryRun?: boolean;
+}
+
+export const POST = apiHandler(async (request: NextRequest) => {
+  const { user } = await requireAdmin();
+  annotate({ action: { name: "admin.drain.cumulative" } });
+
+  const { data: rawBody, error: jsonError } = await safeJson<DrainBody>(request);
+  if (jsonError) return jsonError;
+
+  const body: DrainBody = (rawBody && typeof rawBody === "object") ? rawBody : {};
+  const userId =
+    typeof body.userId === "string" && body.userId.length > 0
+      ? body.userId
+      : undefined;
+  // Default to dryRun = true so a malformed body never silently writes.
+  // Operator must explicitly send `dryRun: false` to commit.
+  const dryRun = body.dryRun === false ? false : true;
+
+  await auditLog("admin.drain.cumulative.start", {
+    userId: user.id,
+    ipAddress: getClientIp(request),
+    details: { targetUserId: userId ?? "ALL", dryRun },
+  });
+
+  const lines: string[] = [];
+  const summary = await drainPerSampleCumulative(prisma, {
+    userId,
+    dryRun,
+    log: (line) => lines.push(line),
+  });
+
+  if (!dryRun) {
+    await auditLog("admin.drain.cumulative.commit", {
+      userId: user.id,
+      ipAddress: getClientIp(request),
+      details: {
+        targetUserId: userId ?? "ALL",
+        ...summary.totals,
+      },
+    });
+  }
+
+  if (summary.totals.bucketsCollapsed === 0 && summary.totals.usersScanned === 0 && userId) {
+    return apiError("User not found", 404);
+  }
+
+  return apiSuccess({
+    dryRun,
+    totals: summary.totals,
+    bucketCount: summary.buckets.length,
+    // Truncate to the first 50 buckets so the response stays
+    // bounded — operators inspect the CLI for the full output.
+    bucketsPreview: summary.buckets.slice(0, 50),
+    logTail: lines.slice(-20),
+  });
+});

src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts

@@ -0,0 +1,154 @@
+import { describe, expect, it } from "vitest";
+
+import {
+  bucketRowsByUserDay,
+  canonicalDailyTimestamp,
+  dayKeyForUserTz,
+  sumBucketValues,
+} from "../drain-per-sample-cumulative";
+
+describe("dayKeyForUserTz", () => {
+  it("anchors the calendar day to the user's IANA zone", () => {
+    // 23:45 NZST on 2026-05-16 is still 11:45 UTC on the same day, but
+    // the user's calendar day in Europe/Berlin would be the previous day
+    // because Berlin is 10-11h behind NZST.
+    const instant = new Date("2026-05-16T11:45:00.000Z");
+    expect(dayKeyForUserTz(instant, "Pacific/Auckland")).toBe("2026-05-16");
+    // Europe/Berlin: same instant resolves to local 13:45 — still 2026-05-16.
+    expect(dayKeyForUserTz(instant, "Europe/Berlin")).toBe("2026-05-16");
+    // America/Los_Angeles (UTC-7 in DST): same instant resolves to 04:45 — 2026-05-16.
+    expect(dayKeyForUserTz(instant, "America/Los_Angeles")).toBe("2026-05-16");
+  });
+
+  it("uses sv-SE Intl formatting so output is ISO-shaped", () => {
+    const instant = new Date("2026-01-01T00:00:00.000Z");
+    expect(dayKeyForUserTz(instant, "UTC")).toBe("2026-01-01");
+    // The Pacific/Auckland clock is 13h ahead → already 2026-01-01 mid-afternoon at UTC midnight.
+    expect(dayKeyForUserTz(instant, "Pacific/Auckland")).toBe("2026-01-01");
+    // Europe/Berlin (UTC+1 in standard time): same instant → 01:00 local → still 2026-01-01.
+    expect(dayKeyForUserTz(instant, "Europe/Berlin")).toBe("2026-01-01");
+  });
+});
+
+describe("canonicalDailyTimestamp", () => {
+  it("anchors a Berlin day-key to the local-noon instant", () => {
+    // 12:00 local in Berlin during DST (CEST, UTC+2) is 10:00 UTC.
+    const ts = canonicalDailyTimestamp("2026-05-16", "Europe/Berlin");
+    expect(ts.toISOString()).toBe("2026-05-16T10:00:00.000Z");
+  });
+
+  it("anchors a UTC day-key to plain 12:00 UTC", () => {
+    const ts = canonicalDailyTimestamp("2026-05-16", "UTC");
+    expect(ts.toISOString()).toBe("2026-05-16T12:00:00.000Z");
+  });
+
+  it("anchors a Pacific/Auckland day-key to the local-noon instant", () => {
+    // 12:00 NZST (UTC+12) → 00:00 UTC on the same day.
+    const ts = canonicalDailyTimestamp("2026-05-16", "Pacific/Auckland");
+    expect(ts.toISOString()).toBe("2026-05-16T00:00:00.000Z");
+  });
+
+  it("handles half-hour-offset zones (India IST = UTC+5:30)", () => {
+    // 12:00 IST → 06:30 UTC on the same day.
+    const ts = canonicalDailyTimestamp("2026-05-16", "Asia/Kolkata");
+    expect(ts.toISOString()).toBe("2026-05-16T06:30:00.000Z");
+  });
+});
+
+describe("bucketRowsByUserDay", () => {
+  it("groups per-sample rows by the user's calendar day", () => {
+    const rows = [
+      {
+        id: "m-1",
+        type: "ACTIVITY_STEPS" as const,
+        value: 1200,
+        measuredAt: new Date("2026-05-16T08:00:00.000Z"), // 10:00 Berlin
+        externalId: "hk-uuid-1",
+      },
+      {
+        id: "m-2",
+        type: "ACTIVITY_STEPS" as const,
+        value: 3400,
+        measuredAt: new Date("2026-05-16T14:00:00.000Z"), // 16:00 Berlin
+        externalId: "hk-uuid-2",
+      },
+      {
+        id: "m-3",
+        type: "ACTIVITY_STEPS" as const,
+        value: 800,
+        measuredAt: new Date("2026-05-17T06:00:00.000Z"), // 08:00 Berlin, next day
+        externalId: "hk-uuid-3",
+      },
+    ];
+    const { byDay } = bucketRowsByUserDay(rows, "Europe/Berlin");
+    expect(byDay.size).toBe(2);
+    expect(byDay.get("2026-05-16")?.length).toBe(2);
+    expect(byDay.get("2026-05-17")?.length).toBe(1);
+  });
+
+  it("skips rows whose externalId is already in stats:... shape (idempotent re-run)", () => {
+    const rows = [
+      {
+        id: "m-1",
+        type: "ACTIVITY_STEPS" as const,
+        value: 5000,
+        measuredAt: new Date("2026-05-16T10:00:00.000Z"),
+        externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16",
+      },
+      {
+        id: "m-2",
+        type: "ACTIVITY_STEPS" as const,
+        value: 1200,
+        measuredAt: new Date("2026-05-16T08:00:00.000Z"),
+        externalId: "hk-uuid-pre-drain",
+      },
+    ];
+    const { byDay } = bucketRowsByUserDay(rows, "Europe/Berlin");
+    // The already-collapsed row is skipped; only the per-sample row
+    // remains in the bucket.
+    expect(byDay.get("2026-05-16")?.length).toBe(1);
+    expect(byDay.get("2026-05-16")?.[0]?.id).toBe("m-2");
+  });
+
+  it("groups rows with NULL externalId (manual entries) like any other per-sample row", () => {
+    const rows = [
+      {
+        id: "m-1",
+        type: "ACTIVITY_STEPS" as const,
+        value: 1200,
+        measuredAt: new Date("2026-05-16T08:00:00.000Z"),
+        externalId: null,
+      },
+    ];
+    const { byDay } = bucketRowsByUserDay(rows, "Europe/Berlin");
+    expect(byDay.get("2026-05-16")?.length).toBe(1);
+  });
+
+  it("returns an empty map when the input is empty", () => {
+    const { byDay } = bucketRowsByUserDay([], "Europe/Berlin");
+    expect(byDay.size).toBe(0);
+  });
+});
+
+describe("sumBucketValues", () => {
+  it("sums a non-empty bucket", () => {
+    const rows = [
+      { id: "1", type: "ACTIVITY_STEPS" as const, value: 1200, measuredAt: new Date(), externalId: null },
+      { id: "2", type: "ACTIVITY_STEPS" as const, value: 3400, measuredAt: new Date(), externalId: null },
+      { id: "3", type: "ACTIVITY_STEPS" as const, value: 800, measuredAt: new Date(), externalId: null },
+    ];
+    expect(sumBucketValues(rows)).toBe(5400);
+  });
+
+  it("returns 0 for an empty bucket", () => {
+    expect(sumBucketValues([])).toBe(0);
+  });
+
+  it("handles fractional values (e.g. ACTIVE_ENERGY_BURNED kcal)", () => {
+    const rows = [
+      { id: "1", type: "ACTIVE_ENERGY_BURNED" as const, value: 12.4, measuredAt: new Date(), externalId: null },
+      { id: "2", type: "ACTIVE_ENERGY_BURNED" as const, value: 7.6, measuredAt: new Date(), externalId: null },
+    ];
+    expect(sumBucketValues(rows)).toBeCloseTo(20.0);
+  });
+});

src/lib/measurements/apple-health-mapping.ts

@@ -345,6 +345,27 @@ export function dailyStatsExternalId(
   return `stats:${hkIdentifier}:${dateYYYYMMDD}`;
 }
 
+/**
+ * v1.4.30 — reverse lookup from a HealthLog `MeasurementType` to the
+ * canonical HealthKit identifier for that type. Used by the drain
+ * script when minting a `dailyStatsExternalId` from a row whose
+ * `hkIdentifier` is not carried on the table (the per-sample ingest
+ * stores only the resolved `MeasurementType`).
+ *
+ * Returns `null` when the type has no HealthKit counterpart (Withings-
+ * only metrics). Callers in the cumulative-drain path can assume the
+ * lookup succeeds because `CUMULATIVE_HK_TYPES` is a subset of the
+ * HK-mapped types.
+ */
+export function hkIdentifierForType(
+  type: MeasurementType,
+): string | null {
+  for (const mapping of Object.values(APPLE_HEALTH_TYPE_MAP)) {
+    if (mapping.measurementType === type) return mapping.hkIdentifier;
+  }
+  return null;
+}
+
 /**
  * HK identifiers the iOS app may emit that HealthLog deliberately does
  * NOT map yet. Listing them here means the batch route can log a

src/lib/measurements/drain-per-sample-cumulative.ts

@@ -0,0 +1,344 @@
+/**
+ * v1.4.30 — drain pre-Option-A per-sample APPLE_HEALTH cumulative rows
+ * into one row per day per type, keyed by the locked `dailyStatsExternalId`
+ * shape. Idempotent: re-running collapses zero rows once every cumulative
+ * bucket already holds a single `stats:...` row.
+ *
+ * Scope per `CUMULATIVE_HK_TYPES`:
+ *   ACTIVITY_STEPS, ACTIVE_ENERGY_BURNED, FLIGHTS_CLIMBED,
+ *   WALKING_RUNNING_DISTANCE, TIME_IN_DAYLIGHT
+ *
+ * Per user × type × calendar day (anchored to `User.timezone`):
+ *   1. SELECT all `Measurement` rows with `source = 'APPLE_HEALTH'` and
+ *      `type = <cumulative type>` and `measuredAt` within that user's
+ *      calendar day boundary and `externalId NOT LIKE 'stats:%'`.
+ *   2. If 0 rows → continue.
+ *   3. If 1 row whose externalId already follows the `stats:...`
+ *      shape → continue (already collapsed).
+ *   4. SUM the values; pick canonical timestamp = midday UTC of the
+ *      user's calendar day (matches the Withings activity-sync
+ *      convention per R-A §5 / W17b).
+ *   5. UPSERT a row with `externalId = dailyStatsExternalId(...)`,
+ *      `value = sumValue`, `measuredAt = canonicalTimestamp`.
+ *   6. DELETE the original per-sample rows in the same transaction.
+ *
+ * Designed to be invoked by both:
+ *   - the CLI at `scripts/drain-per-sample-cumulative.ts`
+ *   - the admin endpoint at
+ *     `POST /api/admin/drain-per-sample-cumulative`
+ *
+ * The dry-run path emits the same per-user-day summary without
+ * touching the DB so the operator can inspect what would change before
+ * committing.
+ */
+import { Prisma } from "@/generated/prisma/client";
+import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+
+import {
+  CUMULATIVE_HK_TYPES,
+  dailyStatsExternalId,
+  hkIdentifierForType,
+} from "./apple-health-mapping";
+
+/** Per-(user, type, day) action summary. */
+export interface DrainBucket {
+  userId: string;
+  type: MeasurementType;
+  /** Calendar-day key in the user's timezone (`YYYY-MM-DD`). */
+  dateKey: string;
+  /** Number of per-sample rows scanned for this bucket. */
+  perSampleCount: number;
+  /** SUM of per-sample values for this bucket (canonical-unit). */
+  sumValue: number;
+  /** ISO-8601 canonical timestamp (midday UTC of the user's calendar day). */
+  canonicalTimestamp: string;
+  /** Resulting `externalId` of the daily-aggregated row. */
+  externalId: string;
+}
+
+export interface DrainSummary {
+  /** Did the run actually write to the DB (false for `dryRun`). */
+  dryRun: boolean;
+  /** Per-user-day buckets the drain rewrote (or would rewrite). */
+  buckets: DrainBucket[];
+  /** Aggregate counts across the run. */
+  totals: {
+    usersScanned: number;
+    bucketsCollapsed: number;
+    perSampleRowsDeleted: number;
+    dailyRowsUpserted: number;
+  };
+}
+
+export interface DrainOptions {
+  /** Limit the drain to a single user. Default = every user. */
+  userId?: string;
+  /** Preview-only mode — no DB writes, no transaction commits. */
+  dryRun?: boolean;
+  /** Logger sink — defaults to `console.log`. */
+  log?: (line: string) => void;
+}
+
+/**
+ * Resolve the user's calendar-day key (`YYYY-MM-DD`) for a given
+ * timestamp + timezone. Reuses the same `sv-SE` Intl formatting choice
+ * the mood-entries path locked in v1.4.25 W7b so iOS-side and
+ * server-side day-keys round-trip byte-identically.
+ */
+export function dayKeyForUserTz(date: Date, tz: string): string {
+  return new Intl.DateTimeFormat("sv-SE", { timeZone: tz }).format(date);
+}
+
+/**
+ * Compute the canonical timestamp for a calendar-day key in the user's
+ * timezone. Returns the JS-Date instant at the user's local 12:00 noon
+ * — matches the Withings activity sync convention (one daily row per
+ * type, anchored to midday so the row sorts cleanly between same-day
+ * spot samples). The string returned by `toISOString()` is UTC.
+ */
+export function canonicalDailyTimestamp(
+  dateKey: string,
+  tz: string,
+): Date {
+  // Compute the UTC offset for noon-local of the given day. We don't
+  // have a lightweight TZ-math library on the server, so the trick is:
+  // build "12:00 UTC of the day", read what wall-clock that shows in
+  // the target zone, then shift by the resulting offset.
+  const utcNoon = new Date(`${dateKey}T12:00:00.000Z`);
+  const fmt = new Intl.DateTimeFormat("en-US", {
+    timeZone: tz,
+    hour12: false,
+    hour: "2-digit",
+    minute: "2-digit",
+    timeZoneName: "shortOffset",
+  });
+  // The shortOffset entry looks like "GMT+2", "GMT-5:30", or "GMT".
+  const parts = fmt.formatToParts(utcNoon);
+  const tzPart = parts.find((p) => p.type === "timeZoneName")?.value ?? "GMT";
+  const match = tzPart.match(/GMT([+-])(\d{1,2})(?::(\d{2}))?/);
+  if (!match) {
+    // UTC — noon UTC is noon local.
+    return utcNoon;
+  }
+  const sign = match[1] === "+" ? 1 : -1;
+  const hours = Number.parseInt(match[2], 10);
+  const minutes = match[3] ? Number.parseInt(match[3], 10) : 0;
+  const offsetMinutes = sign * (hours * 60 + minutes);
+  // utcNoon represents 12:00 UTC. The user's local clock reads
+  // 12:00 + offsetMinutes at that instant. To anchor at local 12:00,
+  // subtract the offset.
+  return new Date(utcNoon.getTime() - offsetMinutes * 60 * 1000);
+}
+
+/**
+ * Group an array of per-sample rows into per-day buckets keyed by the
+ * user's timezone. Exposed for unit testing the bucketing semantics
+ * without booting Prisma.
+ */
+export interface PerSampleRow {
+  id: string;
+  type: MeasurementType;
+  value: number;
+  measuredAt: Date;
+  externalId: string | null;
+}
+
+export interface BucketedRows {
+  /** Map keyed by `YYYY-MM-DD` (user TZ). */
+  byDay: Map<string, PerSampleRow[]>;
+}
+
+export function bucketRowsByUserDay(
+  rows: readonly PerSampleRow[],
+  tz: string,
+): BucketedRows {
+  const byDay = new Map<string, PerSampleRow[]>();
+  for (const row of rows) {
+    // Skip rows already in the daily-stats shape — re-running the
+    // drain on a previously-collapsed bucket is a no-op.
+    if (row.externalId !== null && row.externalId.startsWith("stats:")) {
+      continue;
+    }
+    const key = dayKeyForUserTz(row.measuredAt, tz);
+    const slot = byDay.get(key) ?? [];
+    slot.push(row);
+    byDay.set(key, slot);
+  }
+  return { byDay };
+}
+
+/**
+ * Sum the values in a per-day bucket. Tiny helper; keeps the call
+ * site (transactional drain loop) readable.
+ */
+export function sumBucketValues(rows: readonly PerSampleRow[]): number {
+  let acc = 0;
+  for (const row of rows) acc += row.value;
+  return acc;
+}
+
+/**
+ * Run the drain. Idempotent — re-invocation after a successful drain
+ * reports zero buckets collapsed.
+ *
+ * The function does NOT enforce the admin gate or the operator-confirm
+ * flag — the CLI wrapper + the admin endpoint own those concerns.
+ */
+export async function drainPerSampleCumulative(
+  prismaClient: PrismaClient,
+  options: DrainOptions = {},
+): Promise<DrainSummary> {
+  const dryRun = options.dryRun ?? false;
+  const log = options.log ?? ((line) => console.log(line));
+
+  const users = options.userId
+    ? await prismaClient.user.findMany({
+        where: { id: options.userId },
+        select: { id: true, timezone: true },
+      })
+    : await prismaClient.user.findMany({
+        select: { id: true, timezone: true },
+      });
+
+  const summary: DrainSummary = {
+    dryRun,
+    buckets: [],
+    totals: {
+      usersScanned: users.length,
+      bucketsCollapsed: 0,
+      perSampleRowsDeleted: 0,
+      dailyRowsUpserted: 0,
+    },
+  };
+
+  for (const user of users) {
+    const tz = user.timezone && user.timezone.length > 0 ? user.timezone : "Europe/Berlin";
+    log(`[drain] user=${user.id} tz=${tz}${dryRun ? " (dry-run)" : ""}`);
+
+    for (const type of CUMULATIVE_HK_TYPES) {
+      const hkIdentifier = hkIdentifierForType(type);
+      if (!hkIdentifier) continue;
+
+      const perSampleRows = (await prismaClient.measurement.findMany({
+        where: {
+          userId: user.id,
+          source: "APPLE_HEALTH",
+          type,
+        },
+        select: {
+          id: true,
+          type: true,
+          value: true,
+          measuredAt: true,
+          externalId: true,
+        },
+        orderBy: { measuredAt: "asc" },
+      })) as PerSampleRow[];
+
+      if (perSampleRows.length === 0) continue;
+
+      const { byDay } = bucketRowsByUserDay(perSampleRows, tz);
+
+      for (const [dateKey, dayRows] of byDay) {
+        if (dayRows.length === 0) continue;
+
+        // Already-collapsed buckets are caught above (rows with
+        // externalId starting "stats:" never enter the bucket map).
+        // A single per-sample row left over is still drained — that
+        // way the externalId converges to the canonical shape even
+        // when iOS happened to emit exactly one sample on a quiet
+        // day.
+
+        const sumValue = sumBucketValues(dayRows);
+        const canonicalTs = canonicalDailyTimestamp(dateKey, tz);
+        const externalId = dailyStatsExternalId(hkIdentifier, dateKey);
+
+        const bucket: DrainBucket = {
+          userId: user.id,
+          type,
+          dateKey,
+          perSampleCount: dayRows.length,
+          sumValue,
+          canonicalTimestamp: canonicalTs.toISOString(),
+          externalId,
+        };
+        summary.buckets.push(bucket);
+        summary.totals.bucketsCollapsed += 1;
+
+        if (!dryRun) {
+          await prismaClient.$transaction(async (tx) => {
+            // Upsert the daily-aggregated row first, then drop the
+            // per-sample rows. The unique index
+            // (userId, type, source, externalId) makes the upsert
+            // idempotent across re-runs.
+            await tx.measurement.upsert({
+              where: {
+                userId_type_source_externalId: {
+                  userId: user.id,
+                  type,
+                  source: "APPLE_HEALTH",
+                  externalId,
+                },
+              },
+              create: {
+                userId: user.id,
+                type,
+                value: sumValue,
+                unit: dayRows[0]?.value !== undefined
+                  ? // pick the canonical unit from the existing row by
+                    // looking it up via the mapping table; the per-sample
+                    // rows all carry the same unit on a given type, so
+                    // any of them would do.
+                    await resolveCanonicalUnit(tx, user.id, type)
+                  : "count",
+                source: "APPLE_HEALTH",
+                measuredAt: canonicalTs,
+                externalId,
+              },
+              update: {
+                value: sumValue,
+                measuredAt: canonicalTs,
+              },
+            });
+
+            // Delete the per-sample rows that contributed to the sum.
+            // Using `id IN (...)` is bounded by the per-bucket cap
+            // (the largest cumulative bucket in real data is a
+            // ~1 440-row stepCount day on a phone-only user).
+            const ids = dayRows.map((r) => r.id);
+            const del = await tx.measurement.deleteMany({
+              where: { id: { in: ids } },
+            });
+            summary.totals.perSampleRowsDeleted += del.count;
+            summary.totals.dailyRowsUpserted += 1;
+          });
+        } else {
+          summary.totals.perSampleRowsDeleted += dayRows.length;
+          summary.totals.dailyRowsUpserted += 1;
+        }
+      }
+    }
+  }
+
+  log(
+    `[drain] done — usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted}${dryRun ? " (dry-run)" : ""}`,
+  );
+  return summary;
+}
+
+/**
+ * Pull the canonical unit for a `(userId, type)` pair from an existing
+ * per-sample row. Used during the upsert's `create` branch when we
+ * need to populate `Measurement.unit` for the new aggregated row.
+ */
+async function resolveCanonicalUnit(
+  tx: Prisma.TransactionClient,
+  userId: string,
+  type: MeasurementType,
+): Promise<string> {
+  const row = await tx.measurement.findFirst({
+    where: { userId, type, source: "APPLE_HEALTH" },
+    select: { unit: true },
+  });
+  return row?.unit ?? "count";
+}

tests/integration/drain-per-sample-cumulative.test.ts

@@ -0,0 +1,249 @@
+/**
+ * v1.4.30 — real-Postgres integration coverage for
+ * `drainPerSampleCumulative()`. Mirrors the
+ * `measurements-aggregate-daily` fixture: the testcontainer is hot
+ * before this file loads, so the body just truncates + seeds + drains.
+ *
+ * The drain logic operates against `prisma.$transaction` with an upsert
+ * inside; mocking that surface would re-introduce the v1.4.29 lesson
+ * (mocks hide real-SQL behaviour). The container test pins the
+ * idempotent-re-run contract end-to-end.
+ */
+import { beforeEach, describe, expect, it } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import { drainPerSampleCumulative } from "@/lib/measurements/drain-per-sample-cumulative";
+
+const TEST_USER_ID = "user-drain-cumulative";
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+  await getPrismaClient().user.create({
+    data: {
+      id: TEST_USER_ID,
+      username: "drain-cumulative",
+      email: "drain-cumulative@example.test",
+      timezone: "Europe/Berlin",
+    },
+  });
+});
+
+describe("drainPerSampleCumulative (real Postgres)", () => {
+  it("collapses three per-sample step rows on the same day into one daily row", async () => {
+    const prisma = getPrismaClient();
+    await prisma.measurement.createMany({
+      data: [
+        {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 1200,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T08:00:00.000Z"),
+          externalId: "hk-uuid-1",
+        },
+        {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 3400,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T14:00:00.000Z"),
+          externalId: "hk-uuid-2",
+        },
+        {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 800,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T20:00:00.000Z"),
+          externalId: "hk-uuid-3",
+        },
+      ],
+    });
+
+    const summary = await drainPerSampleCumulative(prisma, {
+      userId: TEST_USER_ID,
+      dryRun: false,
+      log: () => {},
+    });
+
+    expect(summary.dryRun).toBe(false);
+    expect(summary.totals.bucketsCollapsed).toBe(1);
+    expect(summary.totals.perSampleRowsDeleted).toBe(3);
+    expect(summary.totals.dailyRowsUpserted).toBe(1);
+
+    const remaining = await prisma.measurement.findMany({
+      where: {
+        userId: TEST_USER_ID,
+        type: "ACTIVITY_STEPS",
+        source: "APPLE_HEALTH",
+      },
+      orderBy: { measuredAt: "asc" },
+    });
+
+    expect(remaining).toHaveLength(1);
+    expect(remaining[0].value).toBe(5400);
+    expect(remaining[0].externalId).toBe(
+      "stats:HKQuantityTypeIdentifierStepCount:2026-05-16",
+    );
+    // 12:00 Berlin during DST = 10:00 UTC.
+    expect(remaining[0].measuredAt.toISOString()).toBe(
+      "2026-05-16T10:00:00.000Z",
+    );
+  });
+
+  it("is idempotent — a second run after a successful drain is a no-op", async () => {
+    const prisma = getPrismaClient();
+    await prisma.measurement.createMany({
+      data: [
+        {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 1200,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T08:00:00.000Z"),
+          externalId: "hk-uuid-1",
+        },
+        {
+          userId: TEST_USER_ID,
+          type: "ACTIVITY_STEPS",
+          value: 800,
+          unit: "steps",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T14:00:00.000Z"),
+          externalId: "hk-uuid-2",
+        },
+      ],
+    });
+
+    await drainPerSampleCumulative(prisma, {
+      userId: TEST_USER_ID,
+      dryRun: false,
+      log: () => {},
+    });
+    const second = await drainPerSampleCumulative(prisma, {
+      userId: TEST_USER_ID,
+      dryRun: false,
+      log: () => {},
+    });
+
+    expect(second.totals.bucketsCollapsed).toBe(0);
+    expect(second.totals.perSampleRowsDeleted).toBe(0);
+    expect(second.totals.dailyRowsUpserted).toBe(0);
+  });
+
+  it("dry-run reports the plan without touching the DB", async () => {
+    const prisma = getPrismaClient();
+    await prisma.measurement.createMany({
+      data: [
+        {
+          userId: TEST_USER_ID,
+          type: "FLIGHTS_CLIMBED",
+          value: 5,
+          unit: "flights",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T08:00:00.000Z"),
+          externalId: "hk-flights-1",
+        },
+        {
+          userId: TEST_USER_ID,
+          type: "FLIGHTS_CLIMBED",
+          value: 3,
+          unit: "flights",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T14:00:00.000Z"),
+          externalId: "hk-flights-2",
+        },
+      ],
+    });
+
+    const summary = await drainPerSampleCumulative(prisma, {
+      userId: TEST_USER_ID,
+      dryRun: true,
+      log: () => {},
+    });
+
+    expect(summary.dryRun).toBe(true);
+    expect(summary.totals.bucketsCollapsed).toBe(1);
+    expect(summary.buckets[0].sumValue).toBe(8);
+    expect(summary.buckets[0].externalId).toBe(
+      "stats:HKQuantityTypeIdentifierFlightsClimbed:2026-05-16",
+    );
+
+    // DB unchanged.
+    const remaining = await prisma.measurement.findMany({
+      where: { userId: TEST_USER_ID, type: "FLIGHTS_CLIMBED" },
+    });
+    expect(remaining).toHaveLength(2);
+  });
+
+  it("ignores spot metrics (PULSE) — only cumulative types collapse", async () => {
+    const prisma = getPrismaClient();
+    await prisma.measurement.createMany({
+      data: [
+        {
+          userId: TEST_USER_ID,
+          type: "PULSE",
+          value: 64,
+          unit: "bpm",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T08:00:00.000Z"),
+          externalId: "hk-pulse-1",
+        },
+        {
+          userId: TEST_USER_ID,
+          type: "PULSE",
+          value: 72,
+          unit: "bpm",
+          source: "APPLE_HEALTH",
+          measuredAt: new Date("2026-05-16T14:00:00.000Z"),
+          externalId: "hk-pulse-2",
+        },
+      ],
+    });
+
+    const summary = await drainPerSampleCumulative(prisma, {
+      userId: TEST_USER_ID,
+      dryRun: false,
+      log: () => {},
+    });
+
+    expect(summary.totals.bucketsCollapsed).toBe(0);
+
+    const remaining = await prisma.measurement.findMany({
+      where: { userId: TEST_USER_ID, type: "PULSE" },
+    });
+    expect(remaining).toHaveLength(2);
+  });
+
+  it("skips rows already in stats:... shape (re-running after an earlier drain)", async () => {
+    const prisma = getPrismaClient();
+    await prisma.measurement.create({
+      data: {
+        userId: TEST_USER_ID,
+        type: "ACTIVITY_STEPS",
+        value: 5400,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: new Date("2026-05-16T10:00:00.000Z"),
+        externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16",
+      },
+    });
+
+    const summary = await drainPerSampleCumulative(prisma, {
+      userId: TEST_USER_ID,
+      dryRun: false,
+      log: () => {},
+    });
+
+    expect(summary.totals.bucketsCollapsed).toBe(0);
+    const remaining = await prisma.measurement.findMany({
+      where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS" },
+    });
+    expect(remaining).toHaveLength(1);
+    expect(remaining[0].value).toBe(5400);
+  });
+});