mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-10 13:22:14 +00:00
feat(measurements): keep the intraday shape of a cumulative day
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.
This commit is contained in:
1 parent
60c30e1e3a
commit
22d7e8c30c
4 files changed
+562
-9
No files matched your search
@@ -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;
|
||||
@@ -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
|
||||
|
||||
@@ -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<DayWriteOutcome> => {
|
||||
// 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;
|
||||
}
|
||||
|
||||
@@ -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:<HKType>:<day>` 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<number>(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<number>(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<void> {
|
||||
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<number> {
|
||||
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<CumulativeDayProfile[]> {
|
||||
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<string, CumulativeDayProfile>();
|
||||
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<CumulativeDayFold | null> {
|
||||
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);
|
||||
}
|
||||
Reference in new issue
Block a user