From 0e92f96e724a956ff3f8d6d3b3210b3f0e2d8f7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 16 May 2026 11:56:39 +0200 Subject: [PATCH] feat(scripts): drain per-sample cumulative HealthKit rows into daily-aggregated rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 117 ++++++ .../drain-per-sample-cumulative/route.ts | 93 +++++ .../drain-per-sample-cumulative.test.ts | 154 ++++++++ src/lib/measurements/apple-health-mapping.ts | 21 ++ .../drain-per-sample-cumulative.ts | 344 ++++++++++++++++++ .../drain-per-sample-cumulative.test.ts | 249 +++++++++++++ 6 files changed, 978 insertions(+) create mode 100644 scripts/drain-per-sample-cumulative.ts create mode 100644 src/app/api/admin/drain-per-sample-cumulative/route.ts create mode 100644 src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts create mode 100644 src/lib/measurements/drain-per-sample-cumulative.ts create mode 100644 tests/integration/drain-per-sample-cumulative.test.ts diff --git a/scripts/drain-per-sample-cumulative.ts b/scripts/drain-per-sample-cumulative.ts new file mode 100644 index 000000000..a226a01ee --- /dev/null +++ b/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 | --all] [--confirm]", + ); + process.exit(0); + } else { + console.error(`Unknown argument: ${arg}`); + process.exit(2); + } + } + return out; +} + +async function main(): Promise { + const args = parseArgs(process.argv.slice(2)); + + if (!args.userId && !args.all) { + console.error( + "Pick a scope: --user (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); +}); diff --git a/src/app/api/admin/drain-per-sample-cumulative/route.ts b/src/app/api/admin/drain-per-sample-cumulative/route.ts new file mode 100644 index 000000000..d3beec971 --- /dev/null +++ b/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(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), + }); +}); diff --git a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts new file mode 100644 index 000000000..db340f475 --- /dev/null +++ b/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); + }); +}); diff --git a/src/lib/measurements/apple-health-mapping.ts b/src/lib/measurements/apple-health-mapping.ts index f546c5bea..c1141bcf6 100644 --- a/src/lib/measurements/apple-health-mapping.ts +++ b/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 diff --git a/src/lib/measurements/drain-per-sample-cumulative.ts b/src/lib/measurements/drain-per-sample-cumulative.ts new file mode 100644 index 000000000..dcc5a7481 --- /dev/null +++ b/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 = ` 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; +} + +export function bucketRowsByUserDay( + rows: readonly PerSampleRow[], + tz: string, +): BucketedRows { + const byDay = new Map(); + 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 { + 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 { + const row = await tx.measurement.findFirst({ + where: { userId, type, source: "APPLE_HEALTH" }, + select: { unit: true }, + }); + return row?.unit ?? "count"; +} diff --git a/tests/integration/drain-per-sample-cumulative.test.ts b/tests/integration/drain-per-sample-cumulative.test.ts new file mode 100644 index 000000000..aa4632a06 --- /dev/null +++ b/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); + }); +});