From 22d7e8c30cf8f1ffb8e949b1a697002791d72355 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 27 Jul 2026 23:00:19 +0200 Subject: [PATCH] feat(measurements): keep the intraday shape of a cumulative day MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cumulative metric has no meaningful latest reading — a partial day's step count is a fragment, not a value — so the only honest comparison is against the same person's own typical total at the same local hour. That needs an hourly substrate and there was none: the nightly cumulative drain collapses per-sample Apple Health rows into one daily total after the grace window and hard-deletes the originals in the same transaction. Today's curve was derivable; every older day's was already gone. Fold the rows into twenty-four running totals and write them down first, in the same transaction and before the delete. The merge is additive so a day drained twice keeps both halves, and the rows are re-read under the advisory lock the drain already takes so an overlapping tick folds nothing. One table, one nightly range delete to bound it. --- .../migration.sql | 56 +++ prisma/schema.prisma | 67 ++++ .../drain-per-sample-cumulative.ts | 75 +++- .../intraday-cumulative-profile.ts | 373 ++++++++++++++++++ 4 files changed, 562 insertions(+), 9 deletions(-) create mode 100644 prisma/migrations/0283_intraday_cumulative_profile/migration.sql create mode 100644 src/lib/measurements/intraday-cumulative-profile.ts diff --git a/prisma/migrations/0283_intraday_cumulative_profile/migration.sql b/prisma/migrations/0283_intraday_cumulative_profile/migration.sql new file mode 100644 index 000000000..c4481a3d3 --- /dev/null +++ b/prisma/migrations/0283_intraday_cumulative_profile/migration.sql @@ -0,0 +1,56 @@ +-- v1.34.0 — keep the intraday shape of a cumulative day before the fold +-- destroys it. +-- +-- The nightly cumulative drain collapses per-sample Apple Health rows into one +-- daily total after the 36-hour grace window and hard-deletes the originals in +-- the same transaction. That left no substrate at all for a "typical by 21:00" +-- comparison on any day older than the grace window — today's curve was +-- derivable, every older day's was already gone. This table holds the +-- twenty-four running totals the drain now folds out of those rows first. +-- +-- Sidecar only: the canonical `stats:` daily measurement stays the reading of +-- record. Nothing stored here is a measurement. +-- +-- Additive — one table, three indexes, one foreign key. No ALTER on an +-- existing table, no DROP. + +CREATE TABLE IF NOT EXISTS "intraday_cumulative_profiles" ( + "id" TEXT NOT NULL, + "user_id" TEXT NOT NULL, + "type" "measurement_type" NOT NULL, + -- Local calendar day in the user's timezone. VARCHAR so the trailing + -- window read and the retention prune are both lexicographic range scans. + "date_key" VARCHAR(10) NOT NULL, + -- Twenty-four running totals; element h is the value accumulated through + -- the END of local hour h, so the array is non-decreasing. + "hourly_cumulative" DOUBLE PRECISION[], + "day_total" DOUBLE PRECISION NOT NULL, + -- Raw per-sample rows that backed the fold. A two-row day is not a shape. + "sample_count" INTEGER NOT NULL, + -- The zone the day was cut on. A later move leaves profiles on the old + -- clock, and the reader drops them rather than compare mismatched hours. + "timezone" TEXT NOT NULL, + "created_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMPTZ(3) NOT NULL, + + CONSTRAINT "intraday_cumulative_profiles_pkey" PRIMARY KEY ("id") +); + +-- The natural identity: one profile per user x type x local day. Makes the +-- drain's upsert idempotent under a re-run or an overlapping nightly tick. +CREATE UNIQUE INDEX IF NOT EXISTS "intraday_cumulative_profiles_user_type_date_key" + ON "intraday_cumulative_profiles" ("user_id", "type", "date_key"); + +-- The trailing-window read the baseline issues: newest days first. +CREATE INDEX IF NOT EXISTS "intraday_cumulative_profiles_user_type_date_desc_idx" + ON "intraday_cumulative_profiles" ("user_id", "type", "date_key" DESC); + +-- The retention prune sweeps every account at once, so it needs an index that +-- does not lead with the user. +CREATE INDEX IF NOT EXISTS "intraday_cumulative_profiles_date_key_idx" + ON "intraday_cumulative_profiles" ("date_key"); + +ALTER TABLE "intraday_cumulative_profiles" + ADD CONSTRAINT "intraday_cumulative_profiles_user_id_fkey" + FOREIGN KEY ("user_id") REFERENCES "users"("id") + ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 60e1fecf7..4cbf49763 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -770,6 +770,11 @@ model User { // granularity from the measurement write hooks; WEEK/MONTH/YEAR // are enqueued through pg-boss for the worker to fold. measurementRollups MeasurementRollup[] + // v1.34.0 — per-(user, type, local day) cumulative hourly profile. The + // shape the nightly cumulative drain folds out of the per-sample rows + // immediately before deleting them, so a "typical by 21:00" comparison + // has a substrate for days older than the 36-hour grace window. + intradayCumulativeProfiles IntradayCumulativeProfile[] // v0.5.4 ios-coord — per-day dedup ledger for MOOD_REMINDER pushes. // One row per (user, local date) the worker has nudged. Prevents // a second push when the cron rechecks the same hour later in the @@ -1953,6 +1958,68 @@ model MeasurementRollup { @@map("measurement_rollups") } +// ─── Intraday cumulative profiles (v1.34.0) ──────────────── +// +// The shape a cumulative day had, kept before the fold destroys it. +// +// A cumulative metric has no meaningful "latest reading" — a partial +// day's step count is a fragment, not a value — so the only honest +// comparison is against the same person's own typical total AT THE +// SAME LOCAL HOUR. That comparison needs an hourly substrate, and +// there was none: the nightly cumulative drain collapses per-sample +// Apple Health rows into one daily total after the 36-hour grace +// window and hard-deletes the originals in the same transaction. +// Today's curve was derivable; every older day's was already gone. +// +// So the drain now folds the rows it is about to delete into +// twenty-four running totals — one per local hour, index `h` carrying +// the total accumulated through the END of local hour `h` — and +// writes them here first. One compact row per user × type × local +// day, a few thousand rows a year across the supported types. +// +// This is a SIDECAR, never a reading. The canonical `stats:` daily +// total stays the measurement of record; nothing reads a value off +// this table that the measurement rows do not already carry. Days +// still inside the grace window have no row yet and are folded live +// from the raw rows on read, so the current day is always available. +model IntradayCumulativeProfile { + id String @id @default(cuid()) + userId String @map("user_id") + type MeasurementType + /// Local calendar day the profile describes (`YYYY-MM-DD`), in the + /// user's timezone at fold time. Lexicographically ordered, so a + /// trailing-window read and the retention prune are both range scans. + dateKey String @map("date_key") @db.VarChar(10) + /// Twenty-four running totals in canonical units — element `h` is the + /// value accumulated from local midnight through the END of local hour + /// `h`, so the array is non-decreasing and `[23]` equals `dayTotal`. + /// A DST day still has exactly 24 slots: the repeated hour folds into + /// its slot and a skipped hour repeats its predecessor's running total, + /// which keeps every stored day index-comparable with every other. + hourlyCumulative Float[] @map("hourly_cumulative") + /// The day's total — `hourlyCumulative[23]`, denormalised so a coverage + /// check never has to read the array. + dayTotal Float @map("day_total") + /// How many raw per-sample rows backed this profile. A day folded from + /// two rows is not a shape; the baseline reader uses this to refuse such + /// a day rather than let it drag the median. + sampleCount Int @map("sample_count") + /// The IANA zone the day was bucketed in. A user who later moves zones + /// leaves behind profiles cut on the old clock; carrying the zone lets + /// the reader drop the mismatched days instead of silently comparing + /// hours that mean different things. + timezone String + createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) + updatedAt DateTime @updatedAt @map("updated_at") @db.Timestamptz(3) + + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@unique([userId, type, dateKey]) + @@index([userId, type, dateKey(sort: Desc)]) + @@index([dateKey]) + @@map("intraday_cumulative_profiles") +} + // ─── Workouts (v1.4.25 W8d) ──────────────────────────────── // // HKWorkout-aligned typed columns for the analytics hot path + an diff --git a/src/lib/measurements/drain-per-sample-cumulative.ts b/src/lib/measurements/drain-per-sample-cumulative.ts index 81973acc7..aa0f9c91b 100644 --- a/src/lib/measurements/drain-per-sample-cumulative.ts +++ b/src/lib/measurements/drain-per-sample-cumulative.ts @@ -18,7 +18,12 @@ * 5. Otherwise preserve legacy late-increment semantics, explicitly stamp * LEGACY_UNKNOWN, and increment syncVersion; fresh/adopted rows are also * LEGACY_UNKNOWN. - * 6. DELETE the original per-sample rows in the same transaction. + * 6. Fold the day's still-live per-sample rows into 24 hourly running totals + * and merge them into `IntradayCumulativeProfile` (v1.34.0). This runs + * inside the same transaction and BEFORE the delete: the shape of a + * cumulative day is the only substrate a "typical by 21:00" comparison + * can be built from, and step 7 is where it used to be destroyed. + * 7. 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` @@ -56,6 +61,11 @@ import { localStartOfDay, type PerSampleRow, } from "./consolidation-tz"; +import { + foldHourlyCumulative, + mergeIntradayCumulativeProfile, + pruneIntradayCumulativeProfiles, +} from "./intraday-cumulative-profile"; // Re-export the shared timezone day-math primitives + `PerSampleRow` // shape so every established `drain-per-sample-cumulative` import site @@ -128,6 +138,11 @@ export interface DrainSummary { * whole global walk — every other day still collapses. */ daysFailed: number; + /** + * Intraday cumulative profiles dropped past the retention bound. Zero on + * a dry run, and zero on most nights — the bound is a year and a bit. + */ + profilesPruned: number; }; } @@ -202,6 +217,7 @@ export async function drainPerSampleCumulative( perSampleRowsDeleted: 0, dailyRowsUpserted: 0, daysFailed: 0, + profilesPruned: 0, }, }; @@ -249,6 +265,8 @@ export async function drainPerSampleCumulative( reducedValue, dayRows, sourceRowIds, + tz, + dateKey, }): Promise => { // Adopt-in-place canonical-noon resolution + a single P2002 retry, // mirroring `dense-intraday-retention.ts`. Two unique indexes can @@ -323,15 +341,23 @@ export async function drainPerSampleCumulative( AUTHORITATIVE_AGGREGATION_PROVENANCE.has(adoptProvenance); // Re-read contributors under the advisory lock. A serialized loser - // sees zero live rows and therefore performs no canonical update. + // sees zero live rows and therefore performs no canonical update — + // and, below, contributes nothing to the day's stored shape either. + // One read serves both: the sum the canonical row needs and the + // timestamps the hourly fold needs. + const liveRows = await tx.measurement.findMany({ + where: { id: { in: sourceRowIds } }, + select: { + id: true, + type: true, + value: true, + measuredAt: true, + externalId: true, + }, + }); const liveIncrement = eidRow !== null && !protectedAggregate - ? ( - await tx.measurement.findMany({ - where: { id: { in: sourceRowIds } }, - select: { value: true }, - }) - ).reduce((sum, row) => sum + row.value, 0) + ? liveRows.reduce((sum, row) => sum + row.value, 0) : 0; const mergedValue = eidRow !== null ? eidRow.value + liveIncrement : reducedValue; @@ -381,6 +407,23 @@ export async function drainPerSampleCumulative( canonicalRowId = created.id; } + // v1.34.0 — keep the day's SHAPE before the delete below removes the + // only rows it can be derived from. A cumulative day has no + // meaningful latest reading, so the one honest comparison is against + // this person's own typical total at the same local hour; that needs + // an hourly substrate, and this transaction is the last moment it + // exists. Folded from the rows re-read under the advisory lock, so a + // serialised loser folds nothing and a late-arrival drain merges only + // its own new rows. + const shape = foldHourlyCumulative(liveRows, dateKey, tz); + if (shape) { + await mergeIntradayCumulativeProfile(tx, { + userId, + type, + increment: shape, + }); + } + // Hard-delete the per-sample rows that contributed to the sum. // EXCLUDE the adopted canonical row: a per-sample row that fell on // local-noon may be the row we just adopted as the daily total, so @@ -480,8 +523,22 @@ export async function drainPerSampleCumulative( summary.totals.usersScanned = usersScanned; + // v1.34.0 — bound the profile sidecar. One indexed range delete per run, + // across every account, on the same nightly tick that writes the profiles. + // A failure here must never fail the drain: the fold is the load-bearing + // half and a missed prune costs a few hundred bytes until tomorrow. + if (!summary.dryRun) { + try { + summary.totals.profilesPruned = + await pruneIntradayCumulativeProfiles(prismaClient); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + log(`[drain] intraday-profile prune skipped — ${message}`); + } + } + log( - `[drain] done — usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFailed=${summary.totals.daysFailed}${options.dryRun ? " (dry-run)" : ""}`, + `[drain] done — usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFailed=${summary.totals.daysFailed} profilesPruned=${summary.totals.profilesPruned}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; } diff --git a/src/lib/measurements/intraday-cumulative-profile.ts b/src/lib/measurements/intraday-cumulative-profile.ts new file mode 100644 index 000000000..bb0891bea --- /dev/null +++ b/src/lib/measurements/intraday-cumulative-profile.ts @@ -0,0 +1,373 @@ +/** + * v1.34.0 — the intraday shape of a cumulative day, kept before the fold + * destroys it. + * + * A cumulative metric has no meaningful "latest reading": four thousand steps + * is a good day at 09:00 and a poor one at 22:00, and `signals-of-day` excludes + * steps for exactly that reason. The only honest comparison is against the same + * person's own typical total AT THE SAME LOCAL HOUR — and that needs an hourly + * substrate, which did not exist. `drainPerSampleCumulative` collapses the + * per-sample Apple Health rows into one daily total after the 36-hour grace + * window and hard-deletes the originals in the same transaction; Fitbit, + * Withings and Polar write daily totals directly and never had an intraday + * stream at all. + * + * This module owns that substrate, in three parts: + * + * 1. `foldHourlyCumulative` — the PURE fold. Per-sample rows for one local + * day become 24 running totals, element `h` carrying everything + * accumulated through the END of local hour `h`. No IO, so the drain and + * the live read share one definition of what a day's shape is. + * 2. `mergeIntradayCumulativeProfile` — the write the drain calls INSIDE its + * own transaction, immediately before the delete. Keyed on + * `(userId, type, dateKey)` and additive, so a day drained twice keeps + * both halves. + * 3. `readCumulativeDayProfiles` — the trailing-window read the same-time + * baseline issues. Persisted rows for the drained days, a live fold for + * the days still inside the grace window (which have no row yet), so the + * current day is always available. + * + * The table is a SIDECAR. The canonical `stats::` measurement + * stays the reading of record; nothing here is a measurement, and no read path + * takes a value off this table that the measurement rows do not already carry. + */ +import type { + MeasurementType, + Prisma, + PrismaClient, +} from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { shiftDateKey, userDayKey } from "@/lib/tz/format"; + +import { + dayKeyForUserTz, + hourOfDayForUserTz, + localDayWindow, + type PerSampleRow, +} from "./consolidation-tz"; + +/** Hours in a stored profile. Always 24, DST days included — see below. */ +export const PROFILE_HOURS = 24; + +/** + * How long a stored profile survives. + * + * The baseline only ever reads a trailing month, so this is not a data + * requirement — it is the bound that keeps a sidecar from growing forever. A + * year and a bit lets a same-time comparison one day reach across a full + * seasonal cycle without the table ever becoming something an operator has to + * think about: at ~200 bytes a row and four supported types, a decade of one + * account is still under a megabyte. + */ +export const CUMULATIVE_PROFILE_RETENTION_DAYS = 400; + +/** + * How far back a MISSING profile is folded live from raw rows. + * + * Days inside the drain's 36-hour grace window still hold their per-sample + * rows and have no persisted profile yet, so the reader folds them on demand — + * otherwise today and yesterday would be permanently invisible to a feature + * whose whole subject is today. Two days covers the 36-hour window plus the + * gap until the next nightly tick. + * + * It is deliberately NOT larger. Folding an arbitrary historical day live + * would mean an unbounded per-sample read on an account that has never been + * drained, on a request path — and on a drained account it would find nothing + * anyway, because the rows are gone. A day older than this with no profile is + * simply absent, which is the truth. + */ +export const CUMULATIVE_PROFILE_LIVE_FOLD_DAYS = 2; + +/** + * Rows a single live day-fold will read. A dense Apple Watch step day is + * ~1 440 rows; the cap is generous headroom over that and exists so a + * pathological account cannot turn a highlight into an unbounded scan. + */ +const MAX_LIVE_FOLD_ROWS = 5000; + +/** Prefix marking an already-collapsed daily-stats row. */ +const DAILY_STATS_PREFIX = "stats:"; + +/** One local day's cumulative shape, as the pure fold produces it. */ +export interface CumulativeDayFold { + /** Local calendar day (`YYYY-MM-DD`) in `timezone`. */ + dateKey: string; + /** IANA zone the day was cut on. */ + timezone: string; + /** + * 24 running totals — element `h` is everything accumulated from local + * midnight through the END of local hour `h`. Non-decreasing by + * construction; `[23]` equals `dayTotal`. + */ + hourlyCumulative: number[]; + /** The day's total. */ + dayTotal: number; + /** Raw per-sample rows that backed the fold. */ + sampleCount: number; +} + +/** A day profile as a reader resolved it, tagged with where it came from. */ +export interface CumulativeDayProfile extends CumulativeDayFold { + /** Whether this came off the table or was folded live from raw rows. */ + origin: "stored" | "live"; +} + +/** + * Fold one local day's per-sample rows into 24 running totals. + * + * Pure. Rows already collapsed into a `stats:` daily total are skipped — a + * day-sum row would dump the whole day into a single hour and invent a shape + * that never happened, the same trap `loadIntradayPulse` documents at its own + * step-bucket loop. Rows that fall outside `dateKey` are skipped too, so a + * caller may hand over a padded window without corrupting the day. + * + * DST costs nothing here. A fall-back day's repeated local hour folds into its + * one slot, and a spring-forward day's skipped hour simply repeats its + * predecessor's running total. Every stored day therefore has exactly 24 slots + * and is index-comparable with every other, which is the only property the + * same-hour comparison actually needs. + * + * Returns `null` when the day carried no usable per-sample row — an honest + * absence, never a zero-filled profile. A day with genuinely zero steps that + * WAS sampled comes back as a real profile of zeros; a day nobody measured + * comes back as nothing, and the two must not be confused. + */ +export function foldHourlyCumulative( + rows: readonly PerSampleRow[], + dateKey: string, + timezone: string, +): CumulativeDayFold | null { + const perHour = new Array(PROFILE_HOURS).fill(0); + let sampleCount = 0; + + for (const row of rows) { + if (row.externalId?.startsWith(DAILY_STATS_PREFIX)) continue; + if (dayKeyForUserTz(row.measuredAt, timezone) !== dateKey) continue; + const hour = hourOfDayForUserTz(row.measuredAt, timezone); + if (hour < 0 || hour >= PROFILE_HOURS) continue; + perHour[hour] += row.value; + sampleCount += 1; + } + + if (sampleCount === 0) return null; + + const hourlyCumulative = new Array(PROFILE_HOURS); + let running = 0; + for (let h = 0; h < PROFILE_HOURS; h += 1) { + running += perHour[h]; + hourlyCumulative[h] = running; + } + + return { + dateKey, + timezone, + hourlyCumulative, + dayTotal: running, + sampleCount, + }; +} + +/** + * Add one fold onto the stored profile for its day, inside the caller's + * transaction. + * + * The drain calls this immediately before deleting the rows the fold came + * from, so the shape is committed by the same commit that destroys its source + * — there is no window in which the samples are gone and the shape was never + * written. + * + * The merge is ADDITIVE, and it has to be. The drain deletes what it folds, so + * it only ever hands over rows nobody folded before: a day drained at the grace + * cutoff and then drained again a week later after a late watch sync would lose + * its first eleven hours under last-write-wins. Element-wise addition of two + * running totals is exactly the running total of the union, because a cumulative + * sum is linear — so the merged curve is the same one a single fold over all the + * rows would have produced. + * + * The caller must serialise its own reads and writes for the day; the cumulative + * drain does, under the per-bucket advisory lock it already takes. And the + * caller must pass only rows still LIVE at merge time: a re-run or the + * serialised loser of two overlapping ticks sees none, folds nothing, and never + * reaches here. + * + * A stored profile cut on a different timezone is replaced rather than added + * to. After a move the old curve's hours mean something else, and summing the + * two would produce a shape that describes no day that ever happened. + */ +export async function mergeIntradayCumulativeProfile( + tx: Prisma.TransactionClient, + args: { + userId: string; + type: MeasurementType; + increment: CumulativeDayFold; + }, +): Promise { + const { userId, type, increment } = args; + const key = { userId, type, dateKey: increment.dateKey }; + + const existing = await tx.intradayCumulativeProfile.findUnique({ + where: { userId_type_dateKey: key }, + select: { hourlyCumulative: true, sampleCount: true, timezone: true }, + }); + + const reusable = + existing !== null && + existing.timezone === increment.timezone && + existing.hourlyCumulative.length === PROFILE_HOURS; + + const hourlyCumulative = reusable + ? increment.hourlyCumulative.map((v, h) => v + existing.hourlyCumulative[h]) + : increment.hourlyCumulative; + const sampleCount = reusable + ? existing.sampleCount + increment.sampleCount + : increment.sampleCount; + + const data = { + hourlyCumulative, + dayTotal: hourlyCumulative[PROFILE_HOURS - 1], + sampleCount, + timezone: increment.timezone, + }; + + await tx.intradayCumulativeProfile.upsert({ + where: { userId_type_dateKey: key }, + create: { ...key, ...data }, + update: data, + }); +} + +/** + * Drop profiles older than the retention bound, across every account. + * + * `dateKey` is a zero-padded `YYYY-MM-DD` string, so the lexicographic + * comparison IS the chronological one and the delete is an index range scan. + * The cutoff is computed in UTC: it is 400 days back, and no timezone on earth + * moves that boundary by more than a day. + */ +export async function pruneIntradayCumulativeProfiles( + prismaClient: PrismaClient, + now: Date = new Date(), + retentionDays: number = CUMULATIVE_PROFILE_RETENTION_DAYS, +): Promise { + const cutoffKey = shiftDateKey( + now.toISOString().slice(0, 10), + -retentionDays, + ); + const { count } = await prismaClient.intradayCumulativeProfile.deleteMany({ + where: { dateKey: { lt: cutoffKey } }, + }); + return count; +} + +/** + * Read a trailing window of day profiles, newest day first. + * + * Stored rows answer the drained days. The days still inside the grace window + * hold no stored row yet, so each is folded live from its raw per-sample rows — + * bounded to `CUMULATIVE_PROFILE_LIVE_FOLD_DAYS` and to `MAX_LIVE_FOLD_ROWS` + * per day. A day that yields neither is absent from the result; the caller + * counts what it got rather than assuming a length. + * + * Profiles cut on a different timezone than the account's current one are + * dropped. After a move, "by 21:00" means a different moment than it did + * before, and averaging the two would quietly compare unlike hours. + */ +export async function readCumulativeDayProfiles(args: { + userId: string; + type: MeasurementType; + timezone: string; + /** Inclusive newest local day to consider (usually the caller's today). */ + endDateKey: string; + /** How many days back from `endDateKey`, inclusive of it. */ + windowDays: number; + prismaClient?: PrismaClient; +}): Promise { + const { userId, type, timezone, endDateKey, windowDays } = args; + const db = args.prismaClient ?? prisma; + const startDateKey = shiftDateKey(endDateKey, -(windowDays - 1)); + + const stored = await db.intradayCumulativeProfile.findMany({ + where: { + userId, + type, + dateKey: { gte: startDateKey, lte: endDateKey }, + }, + orderBy: { dateKey: "desc" }, + select: { + dateKey: true, + timezone: true, + hourlyCumulative: true, + dayTotal: true, + sampleCount: true, + }, + }); + + const byDay = new Map(); + for (const row of stored) { + // A day cut on another clock is not comparable — drop it rather than + // silently mix two meanings of "by 21:00". + if (row.timezone !== timezone) continue; + if (row.hourlyCumulative.length !== PROFILE_HOURS) continue; + byDay.set(row.dateKey, { + dateKey: row.dateKey, + timezone: row.timezone, + hourlyCumulative: row.hourlyCumulative, + dayTotal: row.dayTotal, + sampleCount: row.sampleCount, + origin: "stored", + }); + } + + // The recent days the drain has not reached yet. Folded live, newest first, + // and only for days actually inside the requested window. + for (let back = 0; back < CUMULATIVE_PROFILE_LIVE_FOLD_DAYS; back += 1) { + const dateKey = shiftDateKey(endDateKey, -back); + if (dateKey < startDateKey) break; + if (byDay.has(dateKey)) continue; + const folded = await foldLiveDay(db, userId, type, dateKey, timezone); + if (folded) byDay.set(dateKey, { ...folded, origin: "live" }); + } + + return [...byDay.values()].sort((a, b) => (a.dateKey < b.dateKey ? 1 : -1)); +} + +/** + * Fold one local day straight from its live per-sample rows. Scoped to the + * exact local-day window (DST-aware, 23/24/25 h) so the row cap is only ever + * spent on rows that are actually inside the day. + */ +async function foldLiveDay( + db: PrismaClient, + userId: string, + type: MeasurementType, + dateKey: string, + timezone: string, +): Promise { + const { dayStart, dayEnd } = localDayWindow(dateKey, timezone); + const rows = await db.measurement.findMany({ + where: { + userId, + type, + deletedAt: null, + measuredAt: { gte: dayStart, lt: dayEnd }, + }, + orderBy: { measuredAt: "asc" }, + take: MAX_LIVE_FOLD_ROWS, + select: { + id: true, + type: true, + value: true, + measuredAt: true, + externalId: true, + }, + }); + return foldHourlyCumulative(rows, dateKey, timezone); +} + +/** + * The caller's local calendar day. Thin wrapper so a consumer of this module + * never has to reach for a second day-key implementation. + */ +export function localDateKey(now: Date, timezone: string): string { + return userDayKey(now, timezone); +}