feat(measurements): add a bounded dense intra-day retention tier for HRV/HR

Daytime heart-rate-variability (SDNN) and heart-rate samples need to keep
their intra-day shape long enough for the stress engine to read them, but
the raw per-sample volume must stay bounded. Add a retention tier that
keeps the last 14 days of raw HRV/HR samples and folds everything older to
one daily-mean row via the proven mean-consolidation drain (same idempotent
stats:<HK>:<day> fold, soft-delete, and rollup recompute).

The two dense-tier types are deliberately EXEMPT from the destructive
daily-mean drain allowlist (HIGH_FREQUENCY_MEAN_TYPES) — that drain would
collapse the intra-day shape and soft-delete the raw rows. A regression test
pins both exclusions so a future edit that adds either type to the drain
allowlist fails loudly.

Migration 0108 adds only the partial index that keeps the retention drain's
discovery and per-user scan cheap; the retention bound lives in code so
tuning it never needs a migration.

prisma/migrations/0108_v1100_dense_intraday_retention/migration.sql

@@ -0,0 +1,54 @@
+-- v1.10.0 — dense intra-day retention tier for daytime HRV / heart-rate.
+--
+-- The Stress engine reads the intra-day SDNN (and daytime heart-rate)
+-- SHAPE across the day, not a single daily mean. Apple Health ships those
+-- as discrete per-sample rows, so the engine needs the raw samples to
+-- survive long enough to read. Two facts make that safe:
+--
+--   1. `HEART_RATE_VARIABILITY` and `PULSE` are NOT in the destructive
+--      daily-mean drain's allowlist (`HIGH_FREQUENCY_MEAN_TYPES` in
+--      src/lib/measurements/apple-health-mapping.ts), so the nightly
+--      mean-consolidation never collapses them — the intra-day shape is
+--      preserved by construction. (PULSE was already excluded for the
+--      correlation/scatter readers; HRV is exempted for the same reason
+--      the Stress proxy needs it.)
+--
+--   2. A BOUNDED retention window contains the volume: per-sample HRV / HR
+--      rows older than the window are folded to one daily-mean `stats:` row
+--      (the same idempotent fold + soft-delete the mean-consolidation uses)
+--      and the raw rows are tombstoned. Within the window the raw samples
+--      stay, so the Stress engine always has its intra-day inputs for the
+--      days it scores.
+--
+-- This migration adds ONLY the partial index that keeps the dense-tier
+-- drain's discovery + per-user scan cheap. No new column, no backfill, no
+-- table rewrite. The retention bound itself lives in application code
+-- (DENSE_INTRADAY_RETENTION_DAYS in
+-- src/lib/measurements/dense-intraday-retention.ts) so tuning it never
+-- needs a migration.
+--
+-- The index is keyed on `(user_id, measured_at)` and partial over the two
+-- dense-tier types on live rows only, so it shrinks toward the in-window
+-- working set as out-of-window samples are folded away — it carries no
+-- steady-state cost once a tenant's history is past the retention window.
+--
+-- Idempotent guard (`IF NOT EXISTS`) makes reruns safe. Forward-only.
+--
+-- Lock note (conscious trade-off): a plain, non-`CONCURRENTLY` build, so
+-- it runs inside Prisma's migration transaction and holds an ACCESS
+-- EXCLUSIVE lock on `measurements` for the build. The partial predicate
+-- (two types, `deleted_at IS NULL`) keeps the index small on a typical
+-- self-host, so the build is fast and the lock is not felt; the largest
+-- multi-year HealthKit tenants should deploy at a low-traffic window. A
+-- `CONCURRENTLY` build cannot run inside the migration transaction and is
+-- not warranted given the partial-index size on the vast majority of
+-- deployments.
+--
+-- Reversibility:
+--   DROP INDEX IF EXISTS "measurements_dense_intraday_retention_idx";
+-- A roll-back loses only the scan acceleration; no data is touched.
+
+CREATE INDEX IF NOT EXISTS "measurements_dense_intraday_retention_idx"
+  ON "measurements" ("user_id", "measured_at")
+  WHERE "type" IN ('HEART_RATE_VARIABILITY', 'PULSE')
+    AND "deleted_at" IS NULL;

prisma/schema.prisma

@@ -736,6 +736,11 @@ model Measurement {
   // migration 0087. It accelerates the v1.5.6 legacy step-consolidation
   // discovery scan. Prisma can't express a partial-predicate index, so
   // it lives in raw migration SQL and is intentionally omitted here.
+  // `measurements_dense_intraday_retention_idx` is a partial index over
+  // `type IN ('HEART_RATE_VARIABILITY','PULSE') AND deleted_at IS NULL`
+  // — see migration 0108. It accelerates the v1.10.0 dense intra-day
+  // retention drain's discovery + per-user scan. Same Prisma-can't-
+  // express-a-partial-predicate reason it is omitted here.
   @@map("measurements")
 }
 

src/lib/jobs/dense-intraday-retention.ts

@@ -0,0 +1,163 @@
+/**
+ * v1.10.0 — pg-boss queue + boot-time converging backfill for the dense
+ * intra-day retention drain of daytime HRV / heart-rate samples.
+ *
+ * Modelled on `mean-consolidation.ts`: a discovery query enqueues one job
+ * per user still holding live per-sample dense-tier rows OLDER than the
+ * retention window, the per-user handler runs `runDenseIntradayRetention`,
+ * and the pass is idempotent across reboots (folded rows are soft-deleted
+ * so they drop off the discovery list).
+ *
+ * Unlike the daily-mean consolidation, this pass scopes to the dense-tier
+ * types (`HEART_RATE_VARIABILITY`, `PULSE`) and keeps the last
+ * `DENSE_INTRADAY_RETENTION_DAYS` of raw per-sample rows so the Stress
+ * engine still has its intra-day SDNN shape; only out-of-window samples
+ * fold to a daily mean.
+ *
+ * The queue name MUST be registered in `allQueues` in
+ * `src/lib/jobs/reminder-worker.ts` so pg-boss provisions it at boot; an
+ * unregistered queue silently never drains.
+ */
+import { prisma } from "@/lib/db";
+import { annotate } from "@/lib/logging/context";
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
+import {
+  runDenseIntradayRetention,
+  DENSE_INTRADAY_RETENTION_TYPES,
+  DENSE_INTRADAY_RETENTION_DAYS,
+} from "@/lib/measurements/dense-intraday-retention";
+
+export const DENSE_INTRADAY_RETENTION_QUEUE = "dense-intraday-retention";
+
+/**
+ * Serial concurrency — the drain walks every out-of-window per-sample
+ * dense-tier row for a user and writes a transaction per day;
+ * concurrency-1 keeps it from crowding the request pool, matching the
+ * mean-consolidation + step-consolidation convention.
+ */
+export const DENSE_INTRADAY_RETENTION_CONCURRENCY = 1;
+
+/**
+ * Daily at 03:50 Europe/Berlin — inside the existing 03:xx maintenance
+ * window, after the daily-mean consolidation that the cumulative + mean
+ * drains run, and before the 04:30 insight pre-generation + 04:45
+ * recovery-score so the rollup tier reflects the fold before the nightly
+ * scores read it.
+ */
+export const DENSE_INTRADAY_RETENTION_CRON = "50 3 * * *";
+
+export interface DenseIntradayRetentionPayload {
+  userId: string;
+  enqueuedAt: string;
+}
+
+/**
+ * Per-user queue handler. Runs the dense-tier retention drain for one
+ * account and returns the summary totals so the worker can log them.
+ */
+export async function runDenseIntradayRetentionForUser(
+  userId: string,
+): Promise<{ daysConsolidated: number; perSampleRowsSoftDeleted: number }> {
+  const summary = await runDenseIntradayRetention(prisma, {
+    userId,
+    log: () => {
+      // Silent inside the queue handler — the worker logs the totals.
+    },
+  });
+  annotate({
+    action: {
+      name: "measurement.dense_intraday.retain",
+      details: {
+        days: summary.totals.daysConsolidated,
+        per_sample_rows_soft_deleted: summary.totals.perSampleRowsSoftDeleted,
+        daily_rows_upserted: summary.totals.dailyRowsUpserted,
+      },
+    },
+  });
+  return {
+    daysConsolidated: summary.totals.daysConsolidated,
+    perSampleRowsSoftDeleted: summary.totals.perSampleRowsSoftDeleted,
+  };
+}
+
+/**
+ * Boot-time discovery. Finds every user holding at least one LIVE
+ * per-sample dense-tier row (an `APPLE_HEALTH` row of a
+ * `DENSE_INTRADAY_RETENTION_TYPES` type whose externalId does NOT start
+ * with the daily-stats prefix, that is not tombstoned, AND whose
+ * `measuredAt` is older than the retention window) and enqueues one
+ * retention job per account.
+ *
+ * Idempotent across reboots: once a user's out-of-window per-sample rows
+ * are soft-deleted, the predicate drops them from the discovery set.
+ * pg-boss `singletonKey` coalesces duplicate sends. Best-effort: errors
+ * are returned through the result value so worker boot never fails because
+ * of a retention miss.
+ */
+export async function enqueueBootTimeDenseIntradayRetention(): Promise<{
+  enqueued: number;
+  skipped: number;
+  error: string | null;
+}> {
+  const boss = getGlobalBoss();
+  if (!boss) {
+    return { enqueued: 0, skipped: 0, error: null };
+  }
+
+  try {
+    // The dense-tier type set is a closed compile-time list of enum
+    // members; splice-free — Prisma binds the `type IN (...)` array as
+    // parameters.
+    const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES);
+    if (types.length === 0) {
+      return { enqueued: 0, skipped: 0, error: null };
+    }
+
+    const windowStart = new Date(
+      Date.now() - DENSE_INTRADAY_RETENTION_DAYS * 24 * 60 * 60 * 1000,
+    );
+
+    const users = await prisma.measurement.findMany({
+      where: {
+        source: "APPLE_HEALTH",
+        type: { in: types },
+        deletedAt: null,
+        NOT: { externalId: { startsWith: "stats:" } },
+        measuredAt: { lt: windowStart },
+      },
+      select: { userId: true },
+      distinct: ["userId"],
+    });
+
+    if (users.length === 0) {
+      return { enqueued: 0, skipped: 0, error: null };
+    }
+
+    let enqueued = 0;
+    let skipped = 0;
+    for (const { userId } of users) {
+      const payload: DenseIntradayRetentionPayload = {
+        userId,
+        enqueuedAt: new Date().toISOString(),
+      };
+      const jobId = await boss.send(DENSE_INTRADAY_RETENTION_QUEUE, payload, {
+        retryLimit: 3,
+        retryDelay: 60,
+        retryBackoff: true,
+        singletonKey: `dense-intraday-retention|${userId}`,
+      });
+      if (jobId) {
+        enqueued += 1;
+      } else {
+        skipped += 1;
+      }
+    }
+    return { enqueued, skipped, error: null };
+  } catch (err) {
+    return {
+      enqueued: 0,
+      skipped: 0,
+      error: err instanceof Error ? err.message : String(err),
+    };
+  }
+}

src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts

@@ -0,0 +1,71 @@
+/**
+ * v1.10.0 — computed scores (WX-E). Dense intra-day retention DRAIN-EXEMPTION
+ * regression guard.
+ *
+ * LOAD-BEARING — a regression here is silent and irreversible.
+ *
+ * The Stress engine reads the intra-day SDNN (and daytime HR) SHAPE across
+ * the day, not a single daily mean. The destructive daily-mean drain
+ * (`consolidateDailyMean`) collapses a day's per-sample rows for every type
+ * in its allowlist (`HIGH_FREQUENCY_MEAN_TYPES`) to ONE daily mean and
+ * SOFT-DELETES the raw rows. If `HEART_RATE_VARIABILITY` (or the daytime
+ * `PULSE` the engine relies on) were ever added to that allowlist, the
+ * intra-day shape Stress needs would be destroyed the next night the drain
+ * ran — and the raw rows would be gone for good.
+ *
+ * This test pins both exclusions so a future edit that adds either type to
+ * the drain's allowlist fails CI loudly. The dense intra-day retention tier
+ * (`dense-intraday-retention.ts`) is the ONLY pass allowed to fold these
+ * types, and it does so behind a bounded window that preserves the recent
+ * intra-day samples.
+ */
+import { describe, expect, it } from "vitest";
+import type { MeasurementType } from "@/generated/prisma/client";
+
+import {
+  CUMULATIVE_HK_TYPES,
+  HIGH_FREQUENCY_MEAN_TYPES,
+} from "../apple-health-mapping";
+import { DENSE_INTRADAY_RETENTION_TYPES } from "../dense-intraday-retention";
+
+describe("dense intra-day retention — drain exemption (WX-E)", () => {
+  it("EXEMPTS HEART_RATE_VARIABILITY from the destructive daily-mean drain", () => {
+    // If this fails, the nightly daily-mean drain will collapse intra-day
+    // SDNN to a single daily mean + soft-delete the raw rows, destroying the
+    // shape the Stress engine reads. The dense-tier retention drain is the
+    // only pass allowed to fold HRV, and only out-of-window samples.
+    expect(
+      HIGH_FREQUENCY_MEAN_TYPES.has(
+        "HEART_RATE_VARIABILITY" as MeasurementType,
+      ),
+    ).toBe(false);
+  });
+
+  it("EXEMPTS PULSE from the destructive daily-mean drain", () => {
+    expect(HIGH_FREQUENCY_MEAN_TYPES.has("PULSE" as MeasurementType)).toBe(
+      false,
+    );
+  });
+
+  it("EXEMPTS both dense-tier types from the cumulative (SUM) drain too", () => {
+    // HRV / HR are spot signals, never cumulative — they must not be in the
+    // SUM drain either, which would also collapse + hard-delete them.
+    for (const type of DENSE_INTRADAY_RETENTION_TYPES) {
+      expect(CUMULATIVE_HK_TYPES.has(type)).toBe(false);
+    }
+  });
+
+  it("scopes the dense tier to exactly HEART_RATE_VARIABILITY + PULSE", () => {
+    expect(Array.from(DENSE_INTRADAY_RETENTION_TYPES).sort()).toEqual(
+      ["HEART_RATE_VARIABILITY", "PULSE"].sort(),
+    );
+  });
+
+  it("keeps the dense tier strictly disjoint from the daily-mean drain", () => {
+    // The two passes must never overlap: the daily-mean drain would destroy
+    // the intra-day shape; the dense tier preserves it behind a window.
+    for (const type of DENSE_INTRADAY_RETENTION_TYPES) {
+      expect(HIGH_FREQUENCY_MEAN_TYPES.has(type)).toBe(false);
+    }
+  });
+});

src/lib/measurements/__tests__/dense-intraday-retention.test.ts

@@ -0,0 +1,168 @@
+/**
+ * v1.10.0 — computed scores (WX-E). Dense intra-day retention drain.
+ *
+ * Pins the retention BOUND: the scan only folds per-sample dense-tier rows
+ * OLDER than the retention window (`measuredAt < now - retentionDays`), so
+ * the recent intra-day shape the Stress engine reads is never collapsed.
+ * Also pins the dense-tier scope (HRV + PULSE), the APPLE_HEALTH source
+ * scope, the MEAN reduction + soft-delete, and the rollup recompute.
+ */
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+const { recomputeBucketsForMeasurement } = vi.hoisted(() => ({
+  recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/rollups/measurement-rollups", () => ({
+  recomputeBucketsForMeasurement,
+}));
+
+import {
+  runDenseIntradayRetention,
+  DENSE_INTRADAY_RETENTION_DAYS,
+  DENSE_INTRADAY_RETENTION_TYPES,
+} from "../dense-intraday-retention";
+import type { PerSampleRow } from "../drain-per-sample-cumulative";
+import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+
+function row(
+  id: string,
+  value: number,
+  iso: string,
+  type: MeasurementType,
+  unit = "ms",
+  externalId: string | null = null,
+): PerSampleRow {
+  return { id, type, value, measuredAt: new Date(iso), externalId, unit };
+}
+
+function buildPrismaMock(rowsByType: Record<string, unknown[]>) {
+  const upsert = vi.fn().mockResolvedValue({});
+  const updateMany = vi.fn().mockResolvedValue({ count: 0 });
+  const findManyMeasurement = vi.fn(
+    async (args: { where: { type: string } }) =>
+      rowsByType[args.where.type] ?? [],
+  );
+  const tx = { measurement: { upsert, updateMany } };
+  return {
+    mock: {
+      user: {
+        findMany: vi
+          .fn()
+          .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]),
+      },
+      measurement: { findMany: findManyMeasurement },
+      $transaction: vi.fn(async (cb: (t: unknown) => Promise<unknown>) =>
+        cb(tx),
+      ),
+    } as unknown as PrismaClient,
+    upsert,
+    updateMany,
+    findManyMeasurement,
+  };
+}
+
+beforeEach(() => {
+  recomputeBucketsForMeasurement.mockClear();
+});
+
+describe("runDenseIntradayRetention — retention bound", () => {
+  it("only scans rows OLDER than the retention window", async () => {
+    const { mock, findManyMeasurement } = buildPrismaMock({});
+    await runDenseIntradayRetention(mock, { log: () => {} });
+
+    // Every scan carries a `measuredAt < cutoff` predicate — the retention
+    // boundary. Without it, in-window intra-day samples would be folded and
+    // the Stress engine would lose its shape.
+    for (const call of findManyMeasurement.mock.calls) {
+      const where = (call[0] as unknown as { where: { measuredAt?: { lt: Date } } })
+        .where;
+      expect(where.measuredAt?.lt).toBeInstanceOf(Date);
+    }
+    // The cutoff is ~DENSE_INTRADAY_RETENTION_DAYS in the past.
+    const firstCutoff = (
+      findManyMeasurement.mock.calls[0]?.[0] as unknown as {
+        where: { measuredAt: { lt: Date } };
+      }
+    ).where.measuredAt.lt.getTime();
+    const expected =
+      Date.now() - DENSE_INTRADAY_RETENTION_DAYS * 24 * 60 * 60 * 1000;
+    // Within a generous 60s of the expected boundary.
+    expect(Math.abs(firstCutoff - expected)).toBeLessThan(60_000);
+  });
+
+  it("folds everything when retentionDays = 0 (no window)", async () => {
+    const { mock, findManyMeasurement } = buildPrismaMock({});
+    await runDenseIntradayRetention(mock, { retentionDays: 0, log: () => {} });
+    for (const call of findManyMeasurement.mock.calls) {
+      const where = (call[0] as unknown as { where: { measuredAt?: { lt: Date } } })
+        .where;
+      // No cutoff predicate → the whole history is in scope.
+      expect(where.measuredAt).toBeUndefined();
+    }
+  });
+
+  it("scans exactly the dense-tier types, source-scoped to APPLE_HEALTH", async () => {
+    const { mock, findManyMeasurement } = buildPrismaMock({});
+    await runDenseIntradayRetention(mock, { log: () => {} });
+
+    const scannedTypes = findManyMeasurement.mock.calls.map(
+      (c) => (c[0] as { where: { type: string } }).where.type,
+    );
+    expect(scannedTypes.sort()).toEqual(
+      Array.from(DENSE_INTRADAY_RETENTION_TYPES).sort(),
+    );
+    for (const call of findManyMeasurement.mock.calls) {
+      const where = (call[0] as unknown as { where: { source: string } }).where;
+      expect(where.source).toBe("APPLE_HEALTH");
+    }
+  });
+});
+
+describe("runDenseIntradayRetention — fold flow", () => {
+  it("upserts the day MEAN and soft-deletes the out-of-window rows", async () => {
+    // Two HRV samples on an out-of-window day.
+    const hrvRows = [
+      row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"),
+      row("b", 60, "2026-05-01T09:00:00.000Z", "HEART_RATE_VARIABILITY"),
+    ];
+    const { mock, upsert, updateMany } = buildPrismaMock({
+      HEART_RATE_VARIABILITY: hrvRows,
+    });
+
+    const summary = await runDenseIntradayRetention(mock, {
+      retentionDays: 0,
+      log: () => {},
+    });
+
+    const upsertArg = upsert.mock.calls[0]?.[0] as {
+      create: { value: number; source: string; type: string };
+    };
+    // value is the MEAN (50), not the sum (100).
+    expect(upsertArg.create.value).toBeCloseTo(50, 6);
+    expect(upsertArg.create.source).toBe("APPLE_HEALTH");
+    expect(upsertArg.create.type).toBe("HEART_RATE_VARIABILITY");
+
+    // soft-delete, never hard delete.
+    const updArg = updateMany.mock.calls[0]?.[0] as {
+      data: { deletedAt: Date };
+    };
+    expect(updArg.data.deletedAt).toBeInstanceOf(Date);
+    expect(summary.totals.daysConsolidated).toBe(1);
+    expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not write on a dry-run", async () => {
+    const { mock, upsert } = buildPrismaMock({
+      HEART_RATE_VARIABILITY: [
+        row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"),
+      ],
+    });
+    await runDenseIntradayRetention(mock, {
+      retentionDays: 0,
+      dryRun: true,
+      log: () => {},
+    });
+    expect(upsert).not.toHaveBeenCalled();
+    expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled();
+  });
+});

src/lib/measurements/dense-intraday-retention.ts

@@ -0,0 +1,254 @@
+/**
+ * v1.10.0 — dense intra-day retention tier for daytime HRV / heart-rate.
+ *
+ * The Stress engine (`src/lib/insights/stress-score.ts`) reads the
+ * intra-day SDNN SHAPE across the day — the spread of a day's
+ * `HEART_RATE_VARIABILITY` (and daytime `PULSE`) samples, not a single
+ * daily mean. Apple Health ships those as discrete per-sample rows, so the
+ * raw samples must survive long enough for the engine to read them.
+ *
+ * Two facts make a bounded raw-retention window the right shape rather than
+ * a new storage paradigm:
+ *
+ *   1. `HEART_RATE_VARIABILITY` and `PULSE` are NOT in the destructive
+ *      daily-mean drain's allowlist (`HIGH_FREQUENCY_MEAN_TYPES` in
+ *      `apple-health-mapping.ts`). The nightly `mean-consolidation` never
+ *      collapses them, so the intra-day shape is preserved by construction
+ *      — the drain-exemption the Stress engine relies on. (PULSE was
+ *      already excluded for the correlation/scatter readers; HRV is
+ *      exempted here for the same reason the Stress proxy needs it.) A
+ *      regression test pins both exclusions.
+ *
+ *   2. A BOUNDED retention window contains the volume the exemption would
+ *      otherwise let grow forever. This pass REUSES the proven
+ *      mean-consolidation drain (same `runConsolidation` base, same
+ *      idempotent `stats:<HK>:<day>` fold, same soft-delete + rollup
+ *      recompute) but with the grace cutoff widened from 36 hours to the
+ *      retention window: per-sample HRV / HR rows OLDER than the window are
+ *      folded to one daily-mean row and the raw rows are tombstoned; rows
+ *      INSIDE the window stay raw so the Stress engine always has its
+ *      intra-day inputs for the days it scores.
+ *
+ * The net effect: the last `DENSE_INTRADAY_RETENTION_DAYS` carry the dense
+ * intra-day samples; everything older folds to the same daily-mean grain
+ * the rest of the high-frequency metrics already use, and the
+ * `MeasurementRollup` DAY/WEEK/MONTH/YEAR tiers keep serving the long
+ * history. No new table, no new value column — the retention bound lives
+ * here in code so tuning it never needs a migration.
+ *
+ * Runs on pg-boss (`DENSE_INTRADAY_RETENTION_QUEUE`), modelled on the
+ * `mean-consolidation` boot-time converging-backfill pattern — the
+ * production standalone image strips `tsx`, so this can never be a CLI.
+ */
+import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+
+import {
+  dailyStatsExternalId,
+  hkIdentifierForType,
+} from "./apple-health-mapping";
+import {
+  runConsolidation,
+  type DayWriteOutcome,
+} from "./consolidation-base";
+import { meanBucketValue } from "./consolidate-daily-mean";
+import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups";
+
+/**
+ * The dense-tier types. These are the daytime intra-day signals the Stress
+ * engine reads as a shape. Both are deliberately EXEMPT from the
+ * destructive daily-mean drain (`HIGH_FREQUENCY_MEAN_TYPES`); the
+ * disjointness from that set is the load-bearing invariant a regression
+ * test pins.
+ */
+export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
+  new Set<MeasurementType>(["HEART_RATE_VARIABILITY", "PULSE"]);
+
+/**
+ * Retention bound (days). Per-sample HRV / HR rows older than this window
+ * fold to one daily-mean row and the raw rows tombstone; rows inside the
+ * window stay raw so the Stress engine has its intra-day inputs.
+ *
+ * 14 days covers the Stress engine's 7-day reference window plus a margin
+ * for late watch syncs and a backfilled gap, while keeping the raw
+ * per-sample volume bounded to roughly two weeks of samples per dense type.
+ */
+export const DENSE_INTRADAY_RETENTION_DAYS = 14;
+
+/** The daily-stats externalId prefix marks an already-collapsed row. */
+const DAILY_STATS_PREFIX = "stats:";
+
+export interface DenseIntradayRetentionSummary {
+  dryRun: boolean;
+  totals: {
+    usersScanned: number;
+    daysConsolidated: number;
+    perSampleRowsSoftDeleted: number;
+    dailyRowsUpserted: number;
+  };
+}
+
+export interface DenseIntradayRetentionOptions {
+  /** Limit the pass to a single user. Default = every user. */
+  userId?: string;
+  /** Preview-only mode — no DB writes. */
+  dryRun?: boolean;
+  /** Logger sink — defaults to `console.log`. */
+  log?: (line: string) => void;
+  /**
+   * Retention window in days. Per-sample rows OLDER than `now() -
+   * retentionDays` are folded; rows inside the window stay raw. Defaults to
+   * `DENSE_INTRADAY_RETENTION_DAYS`. An explicit one-shot can pass `0` to
+   * fold everything (no protection window) — used only by tests / a manual
+   * full-collapse, never by the scheduled pass.
+   */
+  retentionDays?: number;
+}
+
+/**
+ * Run the dense intra-day retention drain. Folds per-sample HRV / HR rows
+ * older than the retention window into one daily-mean `stats:` row per user
+ * × type × day and soft-deletes the raw rows; rows inside the window stay
+ * raw. Idempotent — re-invocation after a successful pass converges to zero
+ * work because the folded rows are soft-deleted and so excluded from the
+ * live scan, and the minted stats row is excluded by the `NOT
+ * startsWith('stats:')` predicate.
+ *
+ * Does NOT enforce any auth gate — the queue handler owns that concern.
+ */
+export async function runDenseIntradayRetention(
+  prismaClient: PrismaClient,
+  options: DenseIntradayRetentionOptions = {},
+): Promise<DenseIntradayRetentionSummary> {
+  const log = options.log ?? ((line) => console.log(line));
+  const retentionDays = options.retentionDays ?? DENSE_INTRADAY_RETENTION_DAYS;
+  // The shared base resolves the cutoff from `cutoffHours`; the retention
+  // window is the same mechanism expressed in days, so the rows newer than
+  // it stay raw exactly like the mean-consolidation grace window keeps
+  // today's syncs raw — only the width differs.
+  const cutoffHours = retentionDays > 0 ? retentionDays * 24 : undefined;
+
+  const summary: DenseIntradayRetentionSummary = {
+    dryRun: options.dryRun ?? false,
+    totals: {
+      usersScanned: 0,
+      daysConsolidated: 0,
+      perSampleRowsSoftDeleted: 0,
+      dailyRowsUpserted: 0,
+    },
+  };
+
+  const { usersScanned } = await runConsolidation<MeasurementType>({
+    prismaClient,
+    options: { ...options, cutoffHours },
+    types: DENSE_INTRADAY_RETENTION_TYPES,
+    hkIdentifierForType,
+    dailyStatsExternalId,
+    statsPrefix: DAILY_STATS_PREFIX,
+    // Mean is the correct out-of-window reduction — the same reducer the
+    // daily-mean consolidation uses for the other high-frequency spot
+    // metrics. Reused verbatim so the two surfaces can never drift.
+    reduce: meanBucketValue,
+    // Live per-sample rows for the type, source-scoped to APPLE_HEALTH so
+    // manual + Withings spot rows survive. The single minted stats row is
+    // excluded by the NOT-startsWith predicate so it is never re-folded;
+    // soft-deleted rows are excluded so a re-run converges. `cutoffAt` is
+    // the retention boundary — rows newer than it (inside the window) are
+    // NOT scanned, so the dense intra-day shape is preserved.
+    buildScanWhere: ({ userId, type, cutoffAt, statsPrefix }) => ({
+      userId,
+      source: "APPLE_HEALTH",
+      type,
+      deletedAt: null,
+      NOT: { externalId: { startsWith: statsPrefix } },
+      ...(cutoffAt ? { measuredAt: { lt: cutoffAt } } : {}),
+    }),
+    scanSelect: {
+      id: true,
+      type: true,
+      value: true,
+      measuredAt: true,
+      externalId: true,
+      unit: true,
+    },
+    writeDay: async ({
+      prismaClient: pc,
+      userId,
+      type,
+      externalId,
+      canonicalTimestamp,
+      reducedValue,
+      dayRows,
+      sourceRowIds,
+    }): Promise<DayWriteOutcome> => {
+      const unit = dayRows[0]?.unit ?? "unknown";
+      let removed = 0;
+      await pc.$transaction(async (tx) => {
+        // Mint / refresh the canonical daily-mean row first. The unique
+        // index (userId, type, source, externalId) makes the upsert
+        // idempotent across re-runs. Built field-by-field (no spread) per
+        // the no-mass-assignment convention.
+        await tx.measurement.upsert({
+          where: {
+            userId_type_source_externalId: {
+              userId,
+              type,
+              source: "APPLE_HEALTH",
+              externalId,
+            },
+          },
+          create: {
+            userId,
+            type,
+            value: reducedValue,
+            unit,
+            source: "APPLE_HEALTH",
+            measuredAt: canonicalTimestamp,
+            externalId,
+          },
+          update: {
+            value: reducedValue,
+            measuredAt: canonicalTimestamp,
+            deletedAt: null,
+          },
+        });
+
+        // Soft-delete the out-of-window per-sample rows in the same
+        // transaction — tombstone, never hard-delete; they remain as an
+        // audit trail and drop off the live read + this pass's re-run
+        // discovery.
+        const del = await tx.measurement.updateMany({
+          where: { id: { in: sourceRowIds }, deletedAt: null },
+          data: { deletedAt: new Date() },
+        });
+        removed = del.count;
+      });
+
+      // Recompute the affected (user, type, day) rollup buckets after the
+      // fold commits — the rollup tier reads only `deleted_at IS NULL`, so
+      // the stale DAY bucket must re-aggregate against the single
+      // consolidated row, exactly as the mean-consolidation drain does.
+      await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp);
+
+      return { kind: "written", sourceRowsRemoved: removed };
+    },
+    recordBucket: ({ dayRows, outcome }) => {
+      summary.totals.daysConsolidated += 1;
+      summary.totals.dailyRowsUpserted += 1;
+      summary.totals.perSampleRowsSoftDeleted +=
+        outcome?.kind === "written" ? outcome.sourceRowsRemoved : dayRows.length;
+    },
+    onUserComplete: ({ userId, tz, dryRun }) => {
+      log(
+        `[dense-intraday-retention] user=${userId} tz=${tz}${dryRun ? " (dry-run)" : ""}`,
+      );
+    },
+  });
+
+  summary.totals.usersScanned = usersScanned;
+
+  log(
+    `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`,
+  );
+
+  return summary;
+}