diff --git a/.planning/ios-coord/v1.8.5-server-to-ios-mean-metric-consolidation.md b/.planning/ios-coord/v1.8.5-server-to-ios-mean-metric-consolidation.md new file mode 100644 index 000000000..a694a1892 --- /dev/null +++ b/.planning/ios-coord/v1.8.5-server-to-ios-mean-metric-consolidation.md @@ -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::` 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. diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index d83ef925c..89054b608 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -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); + } } }, ); diff --git a/src/lib/measurements/__tests__/apple-health-mapping.test.ts b/src/lib/measurements/__tests__/apple-health-mapping.test.ts index b9adf88e7..0c30f9b4d 100644 --- a/src/lib/measurements/__tests__/apple-health-mapping.test.ts +++ b/src/lib/measurements/__tests__/apple-health-mapping.test.ts @@ -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(); diff --git a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts index ff95152e6..33e5f8d36 100644 --- a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts +++ b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts @@ -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 () => { diff --git a/src/lib/measurements/apple-health-mapping.ts b/src/lib/measurements/apple-health-mapping.ts index 9ced6a938..4839bc4c5 100644 --- a/src/lib/measurements/apple-health-mapping.ts +++ b/src/lib/measurements/apple-health-mapping.ts @@ -527,6 +527,17 @@ export const HIGH_FREQUENCY_MEAN_TYPES: ReadonlySet = new Set { 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({