feat(measurements): consolidate gait/mobility HealthKit metrics nightly

Extend the daily-mean consolidation to cover the high-frequency
gait/mobility metrics that previously mapped to no consolidation set and
piled up at sampling granularity: walking asymmetry, walking
double-support, walking steadiness, and walking heart-rate average. The
correct daily reduction for these is the mean, matching the Health-app
display; all four stay disjoint from the cumulative (sum) set.

Fold the daily-mean drain onto the existing nightly cumulative-drain tick
so new high-frequency spot samples no longer accumulate raw between worker
reboots. The two passes share one concurrency-1 cron, keeping the
maintenance window a single sequential walk. Boot-time discovery stays as
the back-fill for accounts that accumulated raw rows before this shipped.

Recompute the affected day's rollup bucket after the consolidation
commits, the same way the ingest path does, so a rollup-covered read
reflects the single consolidated row rather than a stale mean derived from
the now soft-deleted per-sample rows.

Regular heart-rate / pulse stays raw: it is a discrete clinical reading
and feeds correlation and scatter analytics. No schema change.
This commit is contained in:
Marc-André Bombeck committed 2026-06-01 19:54:26 +02:00
1 parent 5771b36899
commit 93790aa043
7 files changed
+284 -4

No files matched your search

@@ -0,0 +1,67 @@
# v1.8.5 — server folds high-frequency mean metrics nightly (informational, no iOS change)
Direction: server → iOS. Status: informational + optional. No breaking
change. iOS needs zero changes for correctness.
## What changed on the server
The server now consolidates the high-frequency HealthKit "spot" metrics
into one canonical daily row per metric per day, server-side, on a nightly
cron — the same treatment cumulative metrics (steps, energy, distance)
already get. The daily reduction is the **MEAN** of the day's samples,
written at local-noon under the existing
`stats:<HKQuantityTypeIdentifier>:<YYYY-MM-DD>` externalId shape, and the
per-sample rows are soft-deleted (tombstoned, pruned later).
v1.8.5 extends the consolidated set to cover the gait/mobility metrics that
previously piled up at sampling granularity:
- `WalkingAsymmetryPercentage` (WALKING_ASYMMETRY)
- `WalkingDoubleSupportPercentage` (WALKING_DOUBLE_SUPPORT)
- `AppleWalkingSteadiness` (WALKING_STEADINESS)
- `WalkingHeartRateAverage` (WALKING_HEART_RATE_AVERAGE)
joining the metrics already folded since v1.7.0: `RespiratoryRate`,
`EnvironmentalAudioExposure`, `HeadphoneAudioExposure`, `WalkingSpeed`,
`WalkingStepLength`.
The fold now runs on the nightly maintenance tick (alongside the cumulative
drain), not just at worker boot. A 36-hour grace window keeps the current
day's in-flight watch syncs raw so the live "today" view stays current.
## What this means for iOS
**Nothing is required.** Keep sending these metrics as per-sample rows with
opaque UUID externalIds exactly as today. The server folds them nightly;
the daily read path averages over the single consolidated row, so the
metric pages and detail history read consistently with no client change.
Existing accumulated per-sample rows are back-consolidated automatically on
the first deploy by the boot-time discovery pass — no migration, no manual
backfill.
## Pulse is unchanged — still raw
Regular heart-rate / `PULSE` is deliberately NOT consolidated. It is a
discrete clinical reading and feeds correlation + scatter analytics, which
read raw PULSE rows. Keep posting PULSE per-sample as today. Its display
already daily-averages via the read-path AVG; storage stays raw.
## Optional future optimisation (not required, not yet scheduled)
If iOS later wants to cut upload volume for these mean metrics, it MAY send
them pre-aggregated as a daily value under the same `stats:` shape it
already uses for cumulative metrics — e.g.
`stats:HKQuantityTypeIdentifierWalkingSpeed:2026-06-01`. The server's batch
route already routes any `stats:*` externalId to the idempotent overwrite
path, so no server change is needed to accept it.
One caveat if iOS adopts this: for a mean metric the daily `stats:` value
must carry the **average**, not the sum. Compute it on-device with
`HKStatisticsCollectionQuery` using the `discreteAverage` option (NOT
`cumulativeSum`, which is correct only for the cumulative metrics). The
server-side nightly fold remains the safety net for any client not on that
contract, so this stays a pure bandwidth optimisation rather than a
correctness dependency.
No `docs/api/openapi.yaml` change — the batch contract is unchanged.
+30
View File
@@ -118,6 +118,10 @@ import {
drainPerSampleCumulative,
DRAIN_CUMULATIVE_CUTOFF_HOURS,
} from "@/lib/measurements/drain-per-sample-cumulative";
import {
consolidateDailyMean,
MEAN_CONSOLIDATION_CUTOFF_HOURS,
} from "@/lib/measurements/consolidate-daily-mean";
import {
STEP_CONSOLIDATION_QUEUE,
STEP_CONSOLIDATION_CONCURRENCY,
@@ -2601,6 +2605,32 @@ export async function startReminderWorker() {
recordError();
workerLog("error", "[drain-cumulative] run failed", err);
}
// v1.8.5 — fold the daily-MEAN drain onto the same nightly tick as
// the cumulative (SUM) drain. The mean drain was previously
// boot-discovery only, so between worker reboots new high-frequency
// spot samples (walking speed, respiratory rate, gait/mobility,
// audio exposure) accumulated raw. Running both passes on one
// concurrency-1 cron keeps the maintenance window a single
// sequential walk and never crowds the request pool. The global
// (no `userId`) signature drains every user; the 36-hour grace
// cutoff keeps today's in-flight watch syncs raw. Boot discovery
// stays as the back-fill for accounts that accumulated raw rows
// before this cron shipped.
try {
const meanSummary = await consolidateDailyMean(getWorkerPrisma(), {
dryRun: false,
cutoffHours: MEAN_CONSOLIDATION_CUTOFF_HOURS,
log: (line) => workerLog("info", line),
});
workerLog(
"info",
`[mean-consolidation] triggeredAt=${job.data.triggeredAt} usersScanned=${meanSummary.totals.usersScanned} daysConsolidated=${meanSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${meanSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${meanSummary.totals.dailyRowsUpserted}`,
);
} catch (err) {
recordError();
workerLog("error", "[mean-consolidation] nightly run failed", err);
}
}
},
);
@@ -648,10 +648,27 @@ describe("HIGH_FREQUENCY_MEAN_TYPES (v1.7.0)", () => {
"AUDIO_EXPOSURE_HEADPHONE",
"WALKING_SPEED",
"WALKING_STEP_LENGTH",
// v1.8.5 — the gait/mobility metrics that previously fell to no
// consolidation set and piled up raw at sampling granularity.
"WALKING_ASYMMETRY",
"WALKING_DOUBLE_SUPPORT",
"WALKING_STEADINESS",
"WALKING_HEART_RATE_AVERAGE",
].sort(),
);
});
it("consolidates the v1.8.5 gait/mobility metrics", () => {
for (const type of [
"WALKING_ASYMMETRY",
"WALKING_DOUBLE_SUPPORT",
"WALKING_STEADINESS",
"WALKING_HEART_RATE_AVERAGE",
] as const) {
expect(HIGH_FREQUENCY_MEAN_TYPES.has(type)).toBe(true);
}
});
it("every mean type resolves to a HealthKit identifier", () => {
for (const type of HIGH_FREQUENCY_MEAN_TYPES) {
expect(hkIdentifierForType(type)).not.toBeNull();
@@ -1,4 +1,11 @@
import { describe, it, expect, vi } from "vitest";
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 {
bucketMeanRows,
@@ -6,7 +13,7 @@ import {
meanBucketValue,
} from "../consolidate-daily-mean";
import type { PerSampleRow } from "../drain-per-sample-cumulative";
import type { PrismaClient } from "@/generated/prisma/client";
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
function row(
id: string,
@@ -14,10 +21,11 @@ function row(
iso: string,
externalId: string | null = null,
unit = "m/s",
type: MeasurementType = "WALKING_SPEED",
): PerSampleRow {
return {
id,
type: "WALKING_SPEED",
type,
value,
measuredAt: new Date(iso),
externalId,
@@ -25,6 +33,10 @@ function row(
};
}
beforeEach(() => {
recomputeBucketsForMeasurement.mockClear();
});
describe("consolidate-daily-mean — reducer + bucketing", () => {
it("computes the arithmetic mean of a per-day bucket", () => {
const rows = [
@@ -141,6 +153,80 @@ describe("consolidateDailyMean — drain flow (mocked Prisma)", () => {
};
expect(updArg.data.deletedAt).toBeInstanceOf(Date);
expect(summary.totals.daysConsolidated).toBe(1);
// T3 — the rollup DAY bucket must be recomputed for the touched
// (user, type, day) so a rollup-covered read does not serve a
// pre-drain mean derived from the now soft-deleted samples.
expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(1);
const [uid, type, measuredAt] =
recomputeBucketsForMeasurement.mock.calls[0];
expect(uid).toBe("user-1");
expect(type).toBe("WALKING_SPEED");
expect(measuredAt).toBeInstanceOf(Date);
});
it("routes the v1.8.5 gait/mobility types through the drain", async () => {
const { mock, upsert } = buildPrismaMock({
WALKING_ASYMMETRY: [
row("a", 2.0, "2026-05-01T08:00:00.000Z", null, "%", "WALKING_ASYMMETRY"),
row("b", 4.0, "2026-05-01T09:00:00.000Z", null, "%", "WALKING_ASYMMETRY"),
],
WALKING_HEART_RATE_AVERAGE: [
row(
"c",
90,
"2026-05-01T08:00:00.000Z",
null,
"count/min",
"WALKING_HEART_RATE_AVERAGE",
),
row(
"d",
110,
"2026-05-01T09:00:00.000Z",
null,
"count/min",
"WALKING_HEART_RATE_AVERAGE",
),
],
});
const summary = await consolidateDailyMean(mock, { log: () => {} });
// One consolidated daily row per type — the day's MEAN, not the sum.
const byType = new Map(
upsert.mock.calls.map((c) => {
const arg = c[0] as {
where: { userId_type_source_externalId: { type: string } };
create: { value: number };
};
return [arg.where.userId_type_source_externalId.type, arg.create.value];
}),
);
expect(byType.get("WALKING_ASYMMETRY")).toBeCloseTo(3.0, 6);
expect(byType.get("WALKING_HEART_RATE_AVERAGE")).toBeCloseTo(100, 6);
expect(summary.totals.daysConsolidated).toBe(2);
// Recompute fires once per consolidated (type, day).
expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(2);
});
it("never scans PULSE even alongside the gait types", async () => {
const { mock, findManyMeasurement } = buildPrismaMock({});
await consolidateDailyMean(mock, { log: () => {} });
const scannedTypes = findManyMeasurement.mock.calls.map(
(c) => (c[0] as { where: { type: string } }).where.type,
);
expect(scannedTypes).toContain("WALKING_ASYMMETRY");
expect(scannedTypes).toContain("WALKING_STEADINESS");
expect(scannedTypes).not.toContain("PULSE");
});
it("does not recompute on a dry-run", async () => {
const { mock } = buildPrismaMock({
WALKING_SPEED: [row("a", 1.0, "2026-05-01T08:00:00.000Z")],
});
await consolidateDailyMean(mock, { dryRun: true, log: () => {} });
expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled();
});
it("passes the grace cutoff into the scan when cutoffHours is set", async () => {
@@ -527,6 +527,17 @@ export const HIGH_FREQUENCY_MEAN_TYPES: ReadonlySet<MeasurementType> = new Set<M
"AUDIO_EXPOSURE_HEADPHONE",
"WALKING_SPEED",
"WALKING_STEP_LENGTH",
// v1.8.5 — the gait/mobility metrics Apple Health emits at sampling
// granularity (asymmetry/double-support per walk, walking heart-rate
// average per walk, steadiness as a near-daily rollup). They previously
// mapped to `latest` and belonged to no consolidation set, so every
// sample piled up raw. MEAN is the correct daily reduction and matches
// the Health-app display; all four stay disjoint from
// `CUMULATIVE_HK_TYPES`.
"WALKING_ASYMMETRY",
"WALKING_DOUBLE_SUPPORT",
"WALKING_STEADINESS",
"WALKING_HEART_RATE_AVERAGE",
]);
/**
+14 -1
View File
@@ -12,7 +12,8 @@
*
* Scope per `HIGH_FREQUENCY_MEAN_TYPES`:
* RESPIRATORY_RATE, AUDIO_EXPOSURE_ENV, AUDIO_EXPOSURE_HEADPHONE,
* WALKING_SPEED, WALKING_STEP_LENGTH
* WALKING_SPEED, WALKING_STEP_LENGTH, WALKING_ASYMMETRY,
* WALKING_DOUBLE_SUPPORT, WALKING_STEADINESS, WALKING_HEART_RATE_AVERAGE
*
* PULSE is NOT in the set — correlation/scatter analytics read raw PULSE
* rows. PULSE keeps raw storage; its display stays daily-averaged via
@@ -58,6 +59,7 @@ import {
type DayWriteOutcome,
} from "./consolidation-base";
import { type PerSampleRow } from "./drain-per-sample-cumulative";
import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups";
/**
* Grace window — rows newer than `now() - cutoffHours` stay raw so
@@ -241,6 +243,17 @@ export async function consolidateDailyMean(
});
removed = del.count;
});
// T3 — recompute the affected (user, type, day) rollup buckets after
// the consolidation commits, the same way the batch route does on
// ingest. Without it, a rollup-covered read could serve a pre-drain
// mean derived from the now soft-deleted per-sample rows: the rollup
// tier reads only `deleted_at IS NULL`, so the stale DAY bucket must
// be re-aggregated against the single consolidated row. The
// canonical timestamp is local-noon of the consolidated day, which
// lands the recompute on the correct UTC day bucket.
await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp);
return { kind: "written", sourceRowsRemoved: removed };
},
recordBucket: ({
@@ -199,6 +199,62 @@ describe("consolidateDailyMean (real Postgres)", () => {
expect(live).toHaveLength(2);
});
it("consolidates a v1.8.5 gait type and recomputes the DAY rollup", async () => {
const prisma = getPrismaClient();
await prisma.measurement.createMany({
data: [
{
userId: TEST_USER_ID,
type: "WALKING_ASYMMETRY",
value: 2.0,
unit: "%",
source: "APPLE_HEALTH",
measuredAt: new Date("2026-05-16T08:00:00.000Z"),
externalId: "hk-asym-1",
},
{
userId: TEST_USER_ID,
type: "WALKING_ASYMMETRY",
value: 4.0,
unit: "%",
source: "APPLE_HEALTH",
measuredAt: new Date("2026-05-16T14:00:00.000Z"),
externalId: "hk-asym-2",
},
],
});
const summary = await consolidateDailyMean(prisma, {
userId: TEST_USER_ID,
dryRun: false,
log: () => {},
});
expect(summary.totals.daysConsolidated).toBe(1);
// One live daily MEAN row carrying 3.0 (the mean of 2 + 4), not 6.
const live = await prisma.measurement.findMany({
where: {
userId: TEST_USER_ID,
type: "WALKING_ASYMMETRY",
deletedAt: null,
},
});
expect(live).toHaveLength(1);
expect(live[0].value).toBeCloseTo(3.0, 6);
// T3 — the DAY rollup bucket reflects the single consolidated row, not
// a pre-drain mean over the now soft-deleted samples.
const dayRollup = await prisma.measurementRollup.findMany({
where: {
userId: TEST_USER_ID,
type: "WALKING_ASYMMETRY",
granularity: "DAY",
},
});
expect(dayRollup).toHaveLength(1);
expect(dayRollup[0].mean).toBeCloseTo(3.0, 6);
});
it("is idempotent — a second run is a no-op", async () => {
const prisma = getPrismaClient();
await prisma.measurement.createMany({