diff --git a/prisma/migrations/0108_v1100_dense_intraday_retention/migration.sql b/prisma/migrations/0108_v1100_dense_intraday_retention/migration.sql new file mode 100644 index 000000000..8d8bcd543 --- /dev/null +++ b/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; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index a8787dec8..f47c76dd0 100644 --- a/prisma/schema.prisma +++ b/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") } diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts new file mode 100644 index 000000000..24fd395eb --- /dev/null +++ b/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), + }; + } +} diff --git a/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts b/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts new file mode 100644 index 000000000..1f9d82faa --- /dev/null +++ b/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); + } + }); +}); diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts new file mode 100644 index 000000000..2f039b49f --- /dev/null +++ b/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) { + 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) => + 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(); + }); +}); diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts new file mode 100644 index 000000000..da61b9f6b --- /dev/null +++ b/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::` 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 = + new Set(["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 { + 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({ + 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 => { + 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; +}