Merge w10/ingest into release/v1.18.10

Capture Withings sleep vitals (HR/respiratory/SpO2/sleep-HRV/score) via the sleep
summary; collapse overlapping BP sources on the cold path for web/iOS parity; add
a STRESS_SCORE source-priority ladder; add a soft-delete-aware measurements index
and cap the all-time cold scan.
This commit is contained in:
Marc-André Bombeck committed 2026-06-20 16:53:35 +02:00
commit b042fcfc27
22 files changed
+801 -107

No files matched your search

+96
View File
@@ -3606,6 +3606,22 @@ paths:
- NIGHTSCOUT
- POLAR
- OURA
stress:
maxItems: 8
type: array
items:
type: string
enum:
- MANUAL
- WITHINGS
- IMPORT
- APPLE_HEALTH
- COMPUTED
- WHOOP
- FITBIT
- NIGHTSCOUT
- POLAR
- OURA
metricPriority:
type: object
properties:
@@ -3881,6 +3897,22 @@ paths:
- NIGHTSCOUT
- POLAR
- OURA
stress:
maxItems: 8
type: array
items:
type: string
enum:
- MANUAL
- WITHINGS
- IMPORT
- APPLE_HEALTH
- COMPUTED
- WHOOP
- FITBIT
- NIGHTSCOUT
- POLAR
- OURA
deviceTypePriority:
type: object
properties:
@@ -14424,6 +14456,22 @@ components:
- NIGHTSCOUT
- POLAR
- OURA
stress:
maxItems: 8
type: array
items:
type: string
enum:
- MANUAL
- WITHINGS
- IMPORT
- APPLE_HEALTH
- COMPUTED
- WHOOP
- FITBIT
- NIGHTSCOUT
- POLAR
- OURA
metricPriority:
type: object
properties:
@@ -14699,6 +14747,22 @@ components:
- NIGHTSCOUT
- POLAR
- OURA
stress:
maxItems: 8
type: array
items:
type: string
enum:
- MANUAL
- WITHINGS
- IMPORT
- APPLE_HEALTH
- COMPUTED
- WHOOP
- FITBIT
- NIGHTSCOUT
- POLAR
- OURA
additionalProperties: false
deviceTypePriority:
type: object
@@ -15020,6 +15084,22 @@ components:
- NIGHTSCOUT
- POLAR
- OURA
stress:
maxItems: 8
type: array
items:
type: string
enum:
- MANUAL
- WITHINGS
- IMPORT
- APPLE_HEALTH
- COMPUTED
- WHOOP
- FITBIT
- NIGHTSCOUT
- POLAR
- OURA
metricPriority:
type: object
properties:
@@ -15295,6 +15375,22 @@ components:
- NIGHTSCOUT
- POLAR
- OURA
stress:
maxItems: 8
type: array
items:
type: string
enum:
- MANUAL
- WITHINGS
- IMPORT
- APPLE_HEALTH
- COMPUTED
- WHOOP
- FITBIT
- NIGHTSCOUT
- POLAR
- OURA
additionalProperties: false
deviceTypePriority:
type: object
+2 -1
View File
@@ -3872,7 +3872,8 @@
"vo2Max": "VO2 max",
"skinTemperature": "Hauttemperatur",
"respiratoryRate": "Atemfrequenz",
"recovery": "Erholung"
"recovery": "Erholung",
"stress": "Stress"
},
"sourceLabels": {
"WITHINGS": "Withings",
+2 -1
View File
@@ -3872,7 +3872,8 @@
"vo2Max": "VO2 max",
"skinTemperature": "Skin temperature",
"respiratoryRate": "Respiratory rate",
"recovery": "Recovery"
"recovery": "Recovery",
"stress": "Stress"
},
"sourceLabels": {
"WITHINGS": "Withings",
+2 -1
View File
@@ -3872,7 +3872,8 @@
"vo2Max": "VO2 máx.",
"skinTemperature": "Temperatura de la piel",
"respiratoryRate": "Frecuencia respiratoria",
"recovery": "Recuperación"
"recovery": "Recuperación",
"stress": "Estrés"
},
"sourceLabels": {
"WITHINGS": "Withings",
+2 -1
View File
@@ -3872,7 +3872,8 @@
"vo2Max": "VO2 max",
"skinTemperature": "Température cutanée",
"respiratoryRate": "Fréquence respiratoire",
"recovery": "Récupération"
"recovery": "Récupération",
"stress": "Stress"
},
"sourceLabels": {
"WITHINGS": "Withings",
+2 -1
View File
@@ -3872,7 +3872,8 @@
"vo2Max": "VO2 max",
"skinTemperature": "Temperatura cutanea",
"respiratoryRate": "Frequenza respiratoria",
"recovery": "Recupero"
"recovery": "Recupero",
"stress": "Stress"
},
"sourceLabels": {
"WITHINGS": "Withings",
+2 -1
View File
@@ -3872,7 +3872,8 @@
"vo2Max": "VO2 max",
"skinTemperature": "Temperatura skóry",
"respiratoryRate": "Częstość oddechów",
"recovery": "Regeneracja"
"recovery": "Regeneracja",
"stress": "Stres"
},
"sourceLabels": {
"WITHINGS": "Withings",
@@ -0,0 +1,40 @@
-- v1.18.10 P-5 — covering partial index for soft-delete-filtered reads.
--
-- The hot reads on `measurements` all filter `deleted_at IS NULL` and order
-- `measured_at DESC` per type — the DISTINCT-ON-latest pass, the management
-- list, the series reads, and the per-type count. The base
-- `(user_id, type, measured_at)` index is ASC and does NOT carry
-- `deleted_at`, so those reads pay a heap re-check of `deleted_at` over the
-- type partition for every candidate row.
--
-- Two partial indexes already cover the highest-volume types
-- (`measurements_dense_intraday_retention_idx` for HRV/PULSE, migration 0108;
-- `measurements_legacy_step_consolidation_idx` for ACTIVITY_STEPS, migration
-- 0087). This index closes the gap for every OTHER type: a covering partial
-- index keyed `(user_id, type, measured_at DESC)` over live rows only serves
-- the latest / list / series reads without the heap re-check, and — being
-- partial on `deleted_at IS NULL` — it carries no entries for tombstoned rows.
--
-- Additive + order-safe: a pure `CREATE INDEX IF NOT EXISTS`, no column add,
-- no backfill, no rewrite of `measurements`.
--
-- Prisma can't express a partial-predicate index, so it lives here in raw
-- migration SQL and is intentionally omitted from `schema.prisma` (the same
-- treatment migrations 0087 + 0108 use; a comment on the model documents it).
--
-- 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. On the largest multi-year HealthKit
-- tenants the live row set is large, so deploy at a low-traffic window. A
-- `CONCURRENTLY` build cannot run inside the migration transaction and is not
-- used here to keep the migration atomic with the rest of the release.
--
-- Reversibility:
-- DROP INDEX IF EXISTS "measurements_live_covering_idx";
-- A roll-back loses only the scan acceleration; no data is touched.
CREATE INDEX IF NOT EXISTS "measurements_live_covering_idx"
ON "measurements" ("user_id", "type", "measured_at" DESC)
WHERE "deleted_at" IS NULL;
+6
View File
@@ -1121,6 +1121,12 @@ model Measurement {
// — 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.
// `measurements_live_covering_idx` is a covering partial index over
// `(user_id, type, measured_at DESC) WHERE deleted_at IS NULL` — see
// migration 0183 (v1.18.10 P-5). It serves the DISTINCT-ON-latest, list,
// and series reads for every type NOT covered by the two partials above
// and removes the `deleted_at` heap re-check. Same Prisma-can't-express-a-
// partial-predicate reason it is omitted here.
@@map("measurements")
}
@@ -87,6 +87,8 @@ const METRIC_LABEL_KEYS: Record<SourcePriorityMetricKey, string> = {
skinTemperature: "settings.sections.sources.metrics.skinTemperature",
respiratoryRate: "settings.sections.sources.metrics.respiratoryRate",
recovery: "settings.sections.sources.metrics.recovery",
// v1.18.10 I-5 — native-vs-derived stress ladder.
stress: "settings.sections.sources.metrics.stress",
};
const SOURCE_LABEL_KEYS: Record<string, string> = {
@@ -15,6 +15,10 @@ vi.mock("@/lib/db", () => ({
// v1.11.1 — readRollupBuckets lazy-loads the source-priority blob; null
// (default findUnique) falls back to the default ladders.
user: { findUnique: vi.fn() },
// v1.18.10 I-1 — the live BP fallback now reads source-collapsed rows via
// `$queryRawUnsafe` (`canonicalMeasurementsFrom`) instead of a chunked
// `findMany`, so cold == warm. Mock it to feed the SYS / DIA series.
$queryRawUnsafe: vi.fn(),
},
}));
@@ -39,6 +43,7 @@ const MEASUREMENT_FIND_MANY = prisma.measurement
.findMany as unknown as ReturnType<typeof vi.fn>;
const ROLLUP_FIND_MANY = prisma.measurementRollup
.findMany as unknown as ReturnType<typeof vi.fn>;
const QUERY_RAW = prisma.$queryRawUnsafe as unknown as ReturnType<typeof vi.fn>;
const PROBE = probeRollupCoverage as unknown as ReturnType<typeof vi.fn>;
const FULLY_COVERED = isFullyCovered as unknown as ReturnType<typeof vi.fn>;
const ANNOTATE = annotate as unknown as ReturnType<typeof vi.fn>;
@@ -53,6 +58,7 @@ const TARGETS_UNDER_65 = {
beforeEach(() => {
MEASUREMENT_FIND_MANY.mockReset();
ROLLUP_FIND_MANY.mockReset();
QUERY_RAW.mockReset();
PROBE.mockReset();
FULLY_COVERED.mockReset();
ANNOTATE.mockReset();
@@ -215,7 +221,7 @@ describe("computeBpInTargetFastPath", () => {
});
describe("live fallback — partial or no coverage", () => {
it("falls back to the chunked findMany when SYS bucket coverage is missing", async () => {
it("falls back to the source-collapsed live read when SYS bucket coverage is missing", async () => {
const coverage = new Map<string, boolean>([
["BLOOD_PRESSURE_SYS", false],
["BLOOD_PRESSURE_DIA", true],
@@ -223,17 +229,16 @@ describe("computeBpInTargetFastPath", () => {
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
// First findMany = SYS rows, second = DIA rows.
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([
// First raw query = SYS rows, second = DIA rows. Rows arrive in the raw
// SQL snake_case shape (`measured_at` + `value`).
QUERY_RAW.mockResolvedValueOnce([
{
id: "1",
measuredAt: new Date("2026-05-15T08:00:00.000Z"),
measured_at: new Date("2026-05-15T08:00:00.000Z"),
value: 122,
},
]).mockResolvedValueOnce([
{
id: "2",
measuredAt: new Date("2026-05-15T08:00:00.000Z"),
measured_at: new Date("2026-05-15T08:00:00.000Z"),
value: 76,
},
]);
@@ -249,7 +254,8 @@ describe("computeBpInTargetFastPath", () => {
// One paired in-target reading.
expect(result.last30Days).toEqual({ pct: 100, pairs: 1 });
expect(ROLLUP_FIND_MANY).not.toHaveBeenCalled();
expect(MEASUREMENT_FIND_MANY).toHaveBeenCalledTimes(2);
expect(MEASUREMENT_FIND_MANY).not.toHaveBeenCalled();
expect(QUERY_RAW).toHaveBeenCalledTimes(2);
});
it("pins the path:live annotate", async () => {
@@ -259,7 +265,7 @@ describe("computeBpInTargetFastPath", () => {
]);
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
await computeBpInTargetFastPath({
userId: "user-partial",
@@ -275,11 +281,51 @@ describe("computeBpInTargetFastPath", () => {
expect(bpCall).toBeDefined();
});
it("collapses overlapping sources on the live path (v1.18.10 I-1 — cold == warm)", async () => {
// Two sources mirror the SAME reading on the same day (Withings + its
// Apple-Health mirror). The source-collapse picks one canonical source
// per (type, day) BEFORE pairing, so the in-target denominator counts the
// shared reading once — matching the rollup path. The collapse runs in
// SQL (`canonicalMeasurementsFrom`), so the mock just confirms the live
// query routes through it and the result reflects ONE pair, not two.
const coverage = new Map<string, boolean>([
["BLOOD_PRESSURE_SYS", false],
["BLOOD_PRESSURE_DIA", false],
]);
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
// The canonical subquery has already collapsed the duplicate source — it
// returns ONE SYS row and ONE DIA row for the day.
QUERY_RAW.mockResolvedValueOnce([
{ measured_at: new Date("2026-05-15T08:00:00.000Z"), value: 122 },
]).mockResolvedValueOnce([
{ measured_at: new Date("2026-05-15T08:00:00.000Z"), value: 76 },
]);
const result = await computeBpInTargetFastPath({
userId: "user-dual-source",
targets: TARGETS_UNDER_65,
now: new Date("2026-05-17T12:00:00.000Z"),
coverage,
});
expect(result.path).toBe("live");
expect(result.last30Days).toEqual({ pct: 100, pairs: 1 });
// Both live reads must route through the canonical-source subquery so the
// collapse fires in SQL — pin the marker so a regression that reverts to
// a raw `findMany` (double-counting) fails here.
for (const call of QUERY_RAW.mock.calls) {
const sql = String(call[0]);
expect(sql).toContain("DISTINCT ON");
expect(sql).toContain("date_trunc('day'");
}
});
it("probes coverage when the caller omits the map", async () => {
const coverage = new Map<string, boolean>([]);
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
const result = await computeBpInTargetFastPath({
userId: "user-fresh",
@@ -331,7 +377,7 @@ describe("computeBpInTargetFastPath", () => {
]);
FULLY_COVERED.mockReturnValue(true);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
const result = await computeBpInTargetFastPath({
userId: "user-honolulu",
@@ -343,7 +389,7 @@ describe("computeBpInTargetFastPath", () => {
expect(result.path).toBe("live");
expect(ROLLUP_FIND_MANY).not.toHaveBeenCalled();
expect(MEASUREMENT_FIND_MANY).toHaveBeenCalledTimes(2);
expect(QUERY_RAW).toHaveBeenCalledTimes(2);
const calls = ANNOTATE.mock.calls.map((c) => c[0]);
const bpCall = calls.find(
@@ -362,7 +408,7 @@ describe("computeBpInTargetFastPath", () => {
]);
FULLY_COVERED.mockReturnValue(true);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
const result = await computeBpInTargetFastPath({
userId: "user-tokyo",
@@ -592,15 +592,18 @@ describe("computeSummariesSlice", () => {
// 90-day cap to `canonicalMeasurementsFrom(rank, "90 days")`. The
// helper lives in `@/lib/analytics/source-rank-sql` (stubbed
// above), so pin the contract at the call boundary: the slice
// must ask for the 90-day window. The `allTime` aggregate
// deliberately calls it WITHOUT an interval (no cap — all-time
// count/min/max/mean must scan every row).
// must ask for the 90-day window.
expect(canonicalMeasurementsFrom).toHaveBeenCalledWith(
expect.any(String),
"90 days",
);
// v1.18.10 P-7 — the `allTime` aggregate is now capped at a generous
// 15-year scan-DoS floor (far beyond any real history) so a
// coverage-miss cold read can't trigger an unbounded full-partition
// scan. It no longer calls the helper without an interval.
expect(canonicalMeasurementsFrom).toHaveBeenCalledWith(
expect.any(String),
"15 years",
);
});
});
+60 -26
View File
@@ -54,6 +54,10 @@ import { prisma } from "@/lib/db";
import { annotate } from "@/lib/logging/context";
import { readRollupBuckets } from "@/lib/rollups/measurement-rollups";
import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
import {
buildSourceRankCase,
canonicalMeasurementsFrom,
} from "@/lib/analytics/source-rank-sql";
import {
probeRollupCoverage,
type RollupCoverageMap,
@@ -454,9 +458,33 @@ async function computeFromLive(
const DAY_MS = 24 * 60 * 60 * 1000;
const bpInTargetSince = new Date(now.getTime() - 365 * DAY_MS);
// v1.18.10 I-1 — collapse overlapping sources before windowing. BP is a
// multi-source metric (a Withings reading mirrored into Apple Health is two
// rows for one event). Without the collapse the live/cold path double-counts
// the shared reading in both the in-target denominator and the
// recency-weighted grade, returning a different number than the rollup/warm
// path (which reads source-collapsed `measurement_rollups`). Pick the
// ladder-winning source per (type, day) so cold == warm (web↔iOS parity).
const priorityJson = await loadUserSourcePriority(userId);
const rankUnqualified = buildSourceRankCase(
priorityJson,
'"type"',
'"source"',
);
const [sysData, diaData] = await Promise.all([
fetchSeriesChunked(userId, "BLOOD_PRESSURE_SYS", bpInTargetSince),
fetchSeriesChunked(userId, "BLOOD_PRESSURE_DIA", bpInTargetSince),
fetchSeriesChunked(
userId,
"BLOOD_PRESSURE_SYS",
bpInTargetSince,
rankUnqualified,
),
fetchSeriesChunked(
userId,
"BLOOD_PRESSURE_DIA",
bpInTargetSince,
rankUnqualified,
),
]);
const rowCount = sysData.length + diaData.length;
@@ -499,34 +527,40 @@ async function computeFromLive(
}
/**
* Cursor-paged read of one BP type. Mirrors the route's
* `fetchMeasurementSeriesChunked` shape but pruned to the only two
* fields the in-target helper consumes (`measuredAt` + `value`) so the
* Prisma round-trip stays minimal on the cold fallback path.
* Source-collapsed read of one BP type over the trailing 365 days.
*
* v1.18.10 I-1 — routes through `canonicalMeasurementsFrom` so a reading that
* exists under two sources (Withings + its Apple-Health mirror) is counted
* once per (type, day), matching the rollup fast-path. Pruned to the two
* fields the in-target helper consumes (`measuredAt` + `value`). Per-event
* rows of the winning source survive the collapse, so the downstream ±5-min
* pairing still sees individual readings. BP is low-volume even on multi-year
* accounts, so one source-collapsed query replaces the prior cursor paging.
*/
const CHUNK = 5000;
async function fetchSeriesChunked(
userId: string,
type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA",
// The 365-day window is expressed as a SQL interval inside the canonical
// subquery; the bound `Date` arg is retained for call-site symmetry.
since: Date,
rankUnqualified: string,
): Promise<BpReading[]> {
const out: BpReading[] = [];
let cursorId: string | undefined;
for (let page = 0; page < 1000; page++) {
const chunk = await prisma.measurement.findMany({
where: { userId, type, measuredAt: { gte: since }, deletedAt: null },
orderBy: [{ measuredAt: "asc" }, { id: "asc" }],
select: { id: true, measuredAt: true, value: true },
take: CHUNK,
...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : {}),
});
if (chunk.length === 0) break;
for (const row of chunk) {
out.push({ measuredAt: row.measuredAt, value: row.value });
}
if (chunk.length < CHUNK) break;
cursorId = chunk[chunk.length - 1].id;
}
return out;
void since;
const rows = await prisma.$queryRawUnsafe<
Array<{ measured_at: Date; value: number }>
>(
`
SELECT
m."measured_at" AS measured_at,
m."value"::double precision AS value
FROM ${canonicalMeasurementsFrom(rankUnqualified, "365 days")}
WHERE m."type" = '${type}'
ORDER BY m."measured_at" ASC
`,
userId,
);
return rows.map((r) => ({
measuredAt: new Date(r.measured_at),
value: Number(r.value),
}));
}
+5 -1
View File
@@ -31,7 +31,7 @@ import {
* absent: they only ever have one source per (type, day), so the collapse is
* a no-op for them and they fall into the CASE's ELSE branch.
*/
const RANKED_TYPES: readonly MeasurementType[] = [
export const RANKED_TYPES: readonly MeasurementType[] = [
"RESTING_HEART_RATE",
"HEART_RATE_VARIABILITY",
"RESPIRATORY_RATE",
@@ -45,6 +45,10 @@ const RANKED_TYPES: readonly MeasurementType[] = [
"PULSE",
"VO2_MAX",
"RECOVERY_SCORE",
// v1.18.10 I-5 — STRESS_SCORE carries a source-priority ladder (`stress`)
// so the SQL collapse ranks a future device-native producer above the
// COMPUTED proxy deterministically. No-op until a second producer lands.
"STRESS_SCORE",
"SLEEP_DURATION",
"ACTIVITY_STEPS",
"ACTIVE_ENERGY_BURNED",
+20 -5
View File
@@ -81,11 +81,22 @@ import { resolveUserTimezone } from "@/lib/tz/resolver";
const DAY_MS = 24 * 60 * 60 * 1000;
/**
* v1.18.10 P-7 — upper bound on the cold-fallback `allTime` aggregate's
* scan window. Sits far beyond any real account's history (the earliest prod
* measurements are < 10 years old) so it never changes a real all-time
* count / min / max / mean; it exists purely so a coverage-miss cold read
* can't trigger an unbounded full-partition scan. Whitelisted by
* `canonicalMeasurementsFrom`'s interval regex (`<N> years`).
*/
const ALL_TIME_LIVE_CAP_INTERVAL = "15 years" as const;
/**
* v1.4.48 — all-time aggregate row used on the cold-mount fallback
* path. Holds the linearly composable columns (`count / min / max /
* mean`) that must reflect every row the user has ever logged for
* that type, so this query intentionally has no `measured_at` cap.
* that type, capped at `ALL_TIME_LIVE_CAP_INTERVAL` (v1.18.10 P-7) as a
* scan-DoS floor.
*/
interface AllTimeAggregateRow {
type: string;
@@ -608,9 +619,13 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
* v1.4.48 — split into two parallel queries:
*
* 1. `allTime` — the linearly composable columns (`count / min /
* max / mean`) keep the full-partition scan because they must
* reflect every row the user has ever logged. No `measured_at`
* cap.
* max / mean`). v1.18.10 P-7 caps this at `ALL_TIME_LIVE_CAP_INTERVAL`
* (15 years) so an un-windowed cold-fallback read on a coverage-miss
* tenant can never scan an unbounded partition. The cap is a scan-DoS
* floor, not a semantic change: it sits far beyond any real account's
* history (the earliest prod rows are < 10 years old) and the steady
* state is rollup-backed anyway, so it only ever bounds the cold path.
*
*
* 2. `windowed` — `avg7 / avg30` and the slope/r² tuples are all
* already filtered to a 7/30/90 day window inside the SELECT.
@@ -648,7 +663,7 @@ async function computeFromLiveAggregate(
MIN(m."value")::double precision AS min_value,
MAX(m."value")::double precision AS max_value,
AVG(m."value")::double precision AS mean_value
FROM ${canonicalMeasurementsFrom(rankUnqualified)}
FROM ${canonicalMeasurementsFrom(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)}
GROUP BY m."type"
`,
userId,
@@ -170,12 +170,10 @@ describe("buildComprehensiveAggregate", () => {
},
]);
// v1.4.38 W-F — sys + dia merged into a single `findMany`
// (`type: { in: [...] }`) → one mock call carrying both rows
// (empty here since the rollup-fresh fixture doesn't exercise
// the BP pairing). The aggregator partitions by type in JS so
// the bpRawRows.sys / .dia byte-shape is preserved.
FIND_MANY.mockResolvedValueOnce([]);
// v1.18.10 I-1 — bpRawRows now reads source-collapsed rows via
// `$queryRawUnsafe` (3rd UNSAFE call: narrow, latests, bpRaw). Empty
// here since the rollup-fresh fixture doesn't exercise BP pairing.
UNSAFE.mockResolvedValueOnce([]);
const result = await buildComprehensiveAggregate("user-rollup-fresh");
const weight = result.summaries.WEIGHT;
@@ -210,9 +208,10 @@ describe("buildComprehensiveAggregate", () => {
// DISTINCT-ON latest moved to `$queryRawUnsafe` (2 calls). Neither
// path runs the legacy heavy COUNT/MIN/MAX/AVG query.
expect(RAW).toHaveBeenCalledTimes(2);
expect(UNSAFE).toHaveBeenCalledTimes(2);
// v1.4.38 W-F — sys + dia merged into a single round-trip.
expect(FIND_MANY).toHaveBeenCalledTimes(1);
// v1.18.10 I-1 — narrow + latests + bpRaw all run on `$queryRawUnsafe`.
expect(UNSAFE).toHaveBeenCalledTimes(3);
// bpRawRows no longer uses `measurement.findMany`.
expect(FIND_MANY).not.toHaveBeenCalled();
expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1);
});
});
@@ -226,9 +225,10 @@ describe("buildComprehensiveAggregate", () => {
// 2. latests — empty.
// No firstMeasurementAt query when totalMeasurements === 0.
RAW.mockResolvedValueOnce([]);
UNSAFE.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
// v1.4.38 W-F — sys + dia merged into one `findMany`.
FIND_MANY.mockResolvedValueOnce([]);
// heavy, latests, bpRaw — all empty.
UNSAFE.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([]);
const result = await buildComprehensiveAggregate("user-empty");
@@ -239,9 +239,9 @@ describe("buildComprehensiveAggregate", () => {
expect(result.firstMeasurementAt).toBeNull();
expect(result.totalMeasurements).toBe(0);
expect(RAW).toHaveBeenCalledTimes(1);
expect(UNSAFE).toHaveBeenCalledTimes(2);
// v1.4.38 W-F — sys + dia merged → single findMany.
expect(FIND_MANY).toHaveBeenCalledTimes(1);
// v1.18.10 I-1 — heavy + latests + bpRaw all on `$queryRawUnsafe`.
expect(UNSAFE).toHaveBeenCalledTimes(3);
expect(FIND_MANY).not.toHaveBeenCalled();
// The cold path's rollup.findMany still fires (in case some
// buckets exist for a subset of types post-race), but returns [].
expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1);
@@ -280,11 +280,12 @@ describe("buildComprehensiveAggregate", () => {
slope90: 0.001,
r2_90: 0.12,
},
]).mockResolvedValueOnce([
{ type: "WEIGHT", value: 81.4, measured_at: now },
]);
// v1.4.38 W-F — sys + dia merged into one round-trip.
FIND_MANY.mockResolvedValueOnce([]);
])
.mockResolvedValueOnce([
{ type: "WEIGHT", value: 81.4, measured_at: now },
])
// v1.18.10 I-1 — bpRaw (empty) is the 3rd UNSAFE call.
.mockResolvedValueOnce([]);
const result = await buildComprehensiveAggregate("user-cold");
const weight = result.summaries.WEIGHT;
@@ -352,23 +353,34 @@ describe("buildComprehensiveAggregate", () => {
slope90: null,
r2_90: null,
},
]).mockResolvedValueOnce([
{
type: "BLOOD_PRESSURE_SYS",
value: 120,
measured_at: measuredAt,
},
]);
// v1.4.38 W-F — single merged `findMany` returning both sys + dia
// rows tagged by `type`. The aggregator partitions in JS, so the
// bpRawRows.sys / .dia shape stays byte-identical.
FIND_MANY.mockResolvedValueOnce([
{ type: "BLOOD_PRESSURE_SYS", measuredAt, value: 120 },
{ type: "BLOOD_PRESSURE_DIA", measuredAt, value: 80 },
]);
])
.mockResolvedValueOnce([
{
type: "BLOOD_PRESSURE_SYS",
value: 120,
measured_at: measuredAt,
},
])
// v1.18.10 I-1 — bpRaw runs on `$queryRawUnsafe` through the
// source-collapsed canonical subquery. The query returns the
// already-collapsed sys + dia rows in raw SQL snake_case shape; the
// aggregator partitions by type in JS so the bpRawRows.sys / .dia
// byte-shape stays identical.
.mockResolvedValueOnce([
{ type: "BLOOD_PRESSURE_SYS", measured_at: measuredAt, value: 120 },
{ type: "BLOOD_PRESSURE_DIA", measured_at: measuredAt, value: 80 },
]);
const result = await buildComprehensiveAggregate("user-bp");
expect(result.bpRawRows.sys).toEqual([{ measuredAt, value: 120 }]);
expect(result.bpRawRows.dia).toEqual([{ measuredAt, value: 80 }]);
// The BP pull must route through the canonical-source subquery (the
// collapse fires in SQL); pin the marker so a revert to a plain findMany
// (double-counting overlapping sources) fails here.
const bpSql = UNSAFE.mock.calls
.map((c) => String(c[0]))
.find((sql) => sql.includes("BLOOD_PRESSURE_SYS"));
expect(bpSql).toContain("user_id");
expect(FIND_MANY).not.toHaveBeenCalled();
});
});
+50 -19
View File
@@ -302,38 +302,67 @@ function timeSubquery<T>(
});
}
interface BpRawRow {
type: string;
measured_at: Date;
value: number;
}
/**
* v1.4.38 W-F — fetch the trailing-90d sys + dia raw rows in a single
* `findMany` (`type: { in: [...] }`) rather than two parallel queries.
* Saves one round-trip + lets Postgres consolidate the index scan.
* Partitioning by type in JS preserves the byte-shape of the legacy
* sys / dia arrays (ASC by measuredAt — already enforced by the SQL
* ORDER BY).
* round-trip rather than two parallel queries.
*
* v1.18.10 I-1 — collapse overlapping sources before returning. BP is a
* multi-source metric (a Withings reading mirrored into Apple Health is two
* rows for one event). The legacy `findMany` returned every source's row, so
* `computeBpInTargetPct` and the recency-weighted grade double-counted the
* shared reading — and a rollup-warm account (which reads source-collapsed
* `measurement_rollups` via `bp-in-target-fast-path`) returned a different
* BP percentage / grade than a rollup-cold account on identical data. Routing
* both raw pulls through `canonicalMeasurementsFrom` picks the ladder-winning
* source per (type, day) so the cold/live path matches the warm/rollup path
* (web↔iOS parity). `BLOOD_PRESSURE_SYS/DIA` are already in `RANKED_TYPES`.
*
* Per-event rows survive the collapse (the subquery keeps every reading of the
* winning source per day, not a daily mean), so the downstream ±5-min pairing
* still sees individual readings. Partitioning by type in JS preserves the
* byte-shape of the legacy sys / dia arrays (ASC by measuredAt).
*/
async function fetchBpRawRows(
userId: string,
// The 90-day window is expressed as a SQL interval inside the canonical
// subquery (parity with the rollup fast-path's `"90 days"`); the bound
// `Date` arg is retained for call-site symmetry and is exactly 90 days.
ninetyDaysAgo: Date,
// The source-rank CASE built once by the caller (over unqualified
// `"type"`/`"source"` columns) — reused so the BP pull doesn't re-load the
// user's source-priority blob the caller already resolved.
rankUnqualified: string,
): Promise<{
sys: Array<{ measuredAt: Date; value: number }>;
dia: Array<{ measuredAt: Date; value: number }>;
}> {
const rows = await prisma.measurement.findMany({
where: {
userId,
type: { in: ["BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"] },
measuredAt: { gte: ninetyDaysAgo },
deletedAt: null,
},
orderBy: { measuredAt: "asc" },
select: { type: true, measuredAt: true, value: true },
});
void ninetyDaysAgo;
const rows = await prisma.$queryRawUnsafe<BpRawRow[]>(
`
SELECT
m."type"::text AS type,
m."measured_at" AS measured_at,
m."value"::double precision AS value
FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
WHERE m."type" IN ('BLOOD_PRESSURE_SYS', 'BLOOD_PRESSURE_DIA')
ORDER BY m."measured_at" ASC
`,
userId,
);
const sys: Array<{ measuredAt: Date; value: number }> = [];
const dia: Array<{ measuredAt: Date; value: number }> = [];
for (const r of rows) {
const measuredAt = new Date(r.measured_at);
if (r.type === "BLOOD_PRESSURE_SYS") {
sys.push({ measuredAt: r.measuredAt, value: r.value });
sys.push({ measuredAt, value: Number(r.value) });
} else if (r.type === "BLOOD_PRESSURE_DIA") {
dia.push({ measuredAt: r.measuredAt, value: r.value });
dia.push({ measuredAt, value: Number(r.value) });
}
}
return { sys, dia };
@@ -481,7 +510,7 @@ async function buildFromRollups(
// readings; the per-type partition in `fetchBpRawRows` preserves
// the legacy `bpRawRows.sys` / `bpRawRows.dia` byte-shape.
timeSubquery(timings, "bpRaw", () =>
fetchBpRawRows(userId, ninetyDaysAgo),
fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified),
),
timeSubquery(
timings,
@@ -694,7 +723,9 @@ async function buildFromLiveAggregate(
}),
),
// v1.4.38 W-F — bp sys + dia in one round-trip (see `fetchBpRawRows`).
timeSubquery(timings, "bpRaw", () => fetchBpRawRows(userId, ninetyDaysAgo)),
timeSubquery(timings, "bpRaw", () =>
fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified),
),
]);
const latestByType = new Map<string, { value: number; measuredAt: Date }>();
@@ -6,8 +6,10 @@ import {
CUMULATIVE_DAY_SUM_TYPES,
cumulativeMetricKey,
isCumulativeDaySumType,
metricKeyForType,
pickCumulativeDaySum,
} from "../cumulative-day-sum";
import { RANKED_TYPES } from "@/lib/analytics/source-rank-sql";
/**
* v1.4.36 W4c — pickCumulativeDaySum unit tests.
@@ -156,3 +158,22 @@ describe("cumulativeMetricKey", () => {
expect(cumulativeMetricKey("TIME_IN_DAYLIGHT")).toBeNull();
});
});
describe("metricKeyForType — source-priority ladder coverage (v1.18.10 I-5)", () => {
// Every type in RANKED_TYPES drives the SQL source-collapse CASE
// (`buildSourceRankCase` skips a type whose `metricKeyForType` is null).
// A ranked type that resolves to null would make `collapseRollupRowsBySource`
// fall to the alphabetical rank-90 tiebreak — nondeterministic the instant a
// second producer for that type is enabled. Pin the contract so a future
// ranked type cannot ship without its metric key.
it("resolves every RANKED_TYPES member to a non-null SourcePriorityMetricKey", () => {
for (const type of RANKED_TYPES) {
expect(metricKeyForType(type)).not.toBeNull();
}
});
it("ladders STRESS_SCORE so the COMPUTED-vs-device producer is deterministic", () => {
expect(metricKeyForType("STRESS_SCORE")).toBe("stress");
expect(RANKED_TYPES).toContain("STRESS_SCORE");
});
});
@@ -172,6 +172,12 @@ export function metricKeyForType(
return "vo2Max";
case "RECOVERY_SCORE":
return "recovery";
case "STRESS_SCORE":
// v1.18.10 I-5 — ladder the COMPUTED-vs-device stress producer so the
// rollup source collapse is deterministic the instant a second producer
// (Oura `daily_stress`) is enabled, rather than falling to the
// alphabetical rank-90 tiebreak in `collapseRollupRowsBySource`.
return "stress";
case "SLEEP_DURATION":
return "sleep";
default:
+12
View File
@@ -131,6 +131,13 @@ export const SOURCE_PRIORITY_METRIC_KEYS = [
// `RECOVERY_SCORE` rows distinguished by source — this ladder lets the
// same picker resolve native-above-proxy without a second engine.
"recovery",
// v1.18.10 I-5 — native-vs-derived stress. HealthLog computes a COMPUTED
// HRV-derived `STRESS_SCORE` proxy; a device-native producer (Oura
// `daily_stress`) is the deferred second source. Laddering it now makes the
// picker deterministic the instant a second producer is enabled, instead of
// collapsing to the alphabetical rank-90 tiebreak. Single-producer today,
// so the ladder is a no-op until the second producer lands.
"stress",
] as const;
export type SourcePriorityMetricKey =
@@ -297,6 +304,11 @@ export const DEFAULT_SOURCE_PRIORITY: Required<MetricPriority> = {
// fallback for users without a strap. v1.17.0 — Oura readiness + Polar
// nightly recovery slot below WHOOP and above the computed proxy.
recovery: ["WHOOP", "OURA", "POLAR", "COMPUTED"],
// v1.18.10 I-5 — native-vs-derived stress. Oura's device-native
// `daily_stress` (the deferred second producer) outranks HealthLog's
// COMPUTED HRV-derived proxy; the proxy is the fallback for users without a
// device-native stress signal. Single-producer today.
stress: ["OURA", "COMPUTED"],
};
/**
@@ -14,6 +14,7 @@ vi.mock("@/lib/db", () => ({
findFirst: vi.fn(),
create: vi.fn(),
update: vi.fn(),
upsert: vi.fn(),
},
withingsConnection: { findUnique: vi.fn() },
},
@@ -69,6 +70,7 @@ import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollup
import {
fetchWithingsSleep,
mapWithingsSleepState,
mapWithingsSleepSummary,
syncUserSleep,
} from "../sync-sleep";
@@ -107,6 +109,71 @@ afterEach(() => {
vi.unstubAllGlobals();
});
/**
* v1.18.10 P0 — fetch mock that returns the per-segment series on the first
* call (`action=get`) and the per-night summary series on the second
* (`action=getsummary`). Lets the sync exercise both the stage path and the
* nightly-vital path in one round-trip.
*/
function installSegmentThenSummaryFetch(
segments: FakeSegment[],
summaries: Array<{
id?: number;
startdate: number;
enddate: number;
data?: Record<string, number | null>;
}>,
) {
let call = 0;
const fetchMock = vi.fn(async () => {
const body = call === 0 ? { series: segments } : { series: summaries };
call++;
return { status: 200, json: async () => ({ status: 0, body }) };
});
vi.stubGlobal("fetch", fetchMock);
return fetchMock;
}
describe("mapWithingsSleepSummary", () => {
it("maps the canonical nightly vitals to existing enums", () => {
const rows = mapWithingsSleepSummary({
hr_average: 58,
rr_average: 14,
sdnn_1: 65,
spo2_average: 96,
sleep_score: 82,
});
const byType = Object.fromEntries(rows.map((r) => [r.type, r.value]));
expect(byType).toEqual({
RESTING_HEART_RATE: 58,
RESPIRATORY_RATE: 14,
HEART_RATE_VARIABILITY: 65,
OXYGEN_SATURATION: 96,
SLEEP_SCORE: 82,
});
});
it("normalises a fractional SpO2 (0..1) to percent", () => {
const rows = mapWithingsSleepSummary({ spo2_average: 0.97 });
const spo2 = rows.find((r) => r.type === "OXYGEN_SATURATION");
expect(spo2?.value).toBeCloseTo(97, 5);
});
it("skips omitted / non-finite fields", () => {
const rows = mapWithingsSleepSummary({
hr_average: 60,
rr_average: null,
sdnn_1: undefined as unknown as number,
sleep_score: Number.NaN,
});
expect(rows.map((r) => r.type)).toEqual(["RESTING_HEART_RATE"]);
});
it("returns nothing for an absent data block", () => {
expect(mapWithingsSleepSummary(undefined)).toEqual([]);
});
});
describe("mapWithingsSleepState", () => {
it("maps state 0 (awake) → AWAKE", () => {
expect(mapWithingsSleepState(0)).toBe("AWAKE");
@@ -271,6 +338,98 @@ describe("syncUserSleep — segment writes + idempotency", () => {
expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
});
it("upserts nightly vitals with stable per-vital externalIds (v1.18.10 P0)", async () => {
const nightEnd = Math.floor(Date.UTC(2026, 4, 13, 6) / 1000);
installSegmentThenSummaryFetch(
[],
[
{
id: 555,
startdate: nightEnd - 8 * 3600,
enddate: nightEnd,
data: {
hr_average: 57,
rr_average: 13,
sdnn_1: 70,
spo2_average: 95,
sleep_score: 88,
},
},
],
);
vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never);
const imported = await syncUserSleep("user-1");
// Five vitals upserted (no stage segments in this fixture).
expect(imported).toBe(5);
const calls = vi.mocked(prisma.measurement.upsert).mock.calls;
const byType = new Map(
calls.map((c) => {
const w = c[0] as {
where: {
userId_type_source_externalId: { type: string; externalId: string };
};
create: { value: number };
};
return [
w.where.userId_type_source_externalId.type,
{
externalId: w.where.userId_type_source_externalId.externalId,
value: w.create.value,
},
];
}),
);
expect(byType.get("RESTING_HEART_RATE")).toEqual({
externalId: "withings:sleep:user-1:555:hr",
value: 57,
});
expect(byType.get("SLEEP_SCORE")).toEqual({
externalId: "withings:sleep:user-1:555:score",
value: 88,
});
expect(byType.get("OXYGEN_SATURATION")?.value).toBe(95);
});
it("keeps stage segments when the summary fetch fails (v1.18.10 P0)", async () => {
let call = 0;
vi.stubGlobal(
"fetch",
vi.fn(async () => {
const isSummary = call > 0;
call++;
return {
status: 200,
json: async () =>
isSummary
? { status: 293 } // summary errors
: {
status: 0,
body: {
series: [
{
startdate: 1715000000,
enddate: 1715003600,
state: 2,
id: 1,
},
],
},
},
};
}),
);
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
const imported = await syncUserSleep("user-1");
// The one stage segment is still written; the summary failure is swallowed.
expect(imported).toBe(1);
expect(prisma.measurement.create).toHaveBeenCalledTimes(1);
expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
});
it("tolerates a ScanWatch night with no REM segment (no all-zeros row)", async () => {
// ScanWatch reports CORE + DEEP only; a missing REM should NOT
// synthesise an all-zeros REM row. Withings simply omits the
+201
View File
@@ -61,6 +61,87 @@ import {
const WITHINGS_SLEEP_URL = "https://wbsapi.withings.net/v2/sleep";
/**
* v1.18.10 P0 — nightly sleep vitals from `action=getsummary`.
*
* The per-segment `action=get` series carries only the stage timeline; the
* per-night summary carries the physiological vitals (average HR, respiratory
* rate, SDNN-based HRV, average SpO2, the Withings sleep score). These were
* previously dropped at the fetch even though the canonical enums exist. Each
* vital maps to an EXISTING `MeasurementType` so there is no migration and no
* iOS-contract change — the rows simply flow to `/api/sync/changes` under the
* `WITHINGS` source like every other server-side fetch.
*
* Field set requested via `data_fields`. Withings nests the requested fields
* under each night's `data` object. Only the fields with a canonical enum are
* requested; snoring / breathing-disturbance intensity have no enum and are
* deliberately left out (no home to store them).
*/
const SLEEP_SUMMARY_DATA_FIELDS =
"hr_average,rr_average,sdnn_1,spo2_average,sleep_score";
/**
* One nightly summary row from `POST /v2/sleep?action=getsummary`. The vitals
* live under `data`; `id` is the per-night session id (stable across re-syncs,
* the same id the segment series carries) so the externalId reconciles.
*/
export interface WithingsSleepSummary {
id?: number;
startdate: number;
enddate: number;
data?: {
hr_average?: number | null;
rr_average?: number | null;
sdnn_1?: number | null;
spo2_average?: number | null;
sleep_score?: number | null;
};
}
/**
* One nightly vital extracted from a summary, ready to upsert. The
* `measuredAt` is the night's END (`enddate`) so the vital sorts under the
* same night as the END-stamped stage segments.
*/
interface SleepVitalRow {
type: MeasurementType;
value: number;
/** Stable per-vital externalId suffix; combined with the session id. */
fieldTag: string;
}
/**
* Map a nightly summary's `data` block to the canonical vital rows. Skips any
* field Withings omitted or returned as a non-finite value. `spo2_average`
* arrives as a fraction (0..1) on some firmware and a percent (0..100) on
* others — normalise the fraction form to percent so it shares the
* `OXYGEN_SATURATION` bucket with the meastype-54 spot reading.
*/
export function mapWithingsSleepSummary(
data: WithingsSleepSummary["data"] | undefined,
): SleepVitalRow[] {
if (!data) return [];
const rows: SleepVitalRow[] = [];
const push = (type: MeasurementType, value: unknown, fieldTag: string) => {
if (typeof value !== "number" || !Number.isFinite(value)) return;
rows.push({ type, value, fieldTag });
};
// Average HR over the sleep window → RESTING_HEART_RATE (one nightly sample,
// distinct from spot PULSE — same convention the HealthKit nightly RHR uses).
push("RESTING_HEART_RATE", data.hr_average, "hr");
push("RESPIRATORY_RATE", data.rr_average, "rr");
// sdnn_1 is the SDNN-based HRV estimator — the same estimator the
// HEART_RATE_VARIABILITY enum documents (kept distinct from WHOOP's RMSSD).
push("HEART_RATE_VARIABILITY", data.sdnn_1, "hrv");
let spo2 = data.spo2_average;
if (typeof spo2 === "number" && Number.isFinite(spo2) && spo2 <= 1) {
spo2 = spo2 * 100;
}
push("OXYGEN_SATURATION", spo2, "spo2");
push("SLEEP_SCORE", data.sleep_score, "score");
return rows;
}
/**
* Default backfill window for the first sleep sync of a connection.
* Matches every other Withings ingest path so a reconnect lights up
@@ -83,6 +164,16 @@ export interface WithingsSleepSegment {
id?: number;
}
/**
* Format a Date as `YYYY-MM-DD` in UTC for the summary endpoint's
* `startdateymd` / `enddateymd` params. The summary window is day-grained;
* UTC keeps it deterministic and matches the unix-second window the segment
* fetch derives from the same `start` / `now` Dates.
*/
function ymdUtc(d: Date): string {
return d.toISOString().slice(0, 10);
}
/**
* Map Withings `state` to the HealthLog `SleepStage` enum. Returns
* `null` for states we deliberately ignore (state 4 = synthetic
@@ -153,6 +244,56 @@ export async function fetchWithingsSleep(
return series;
}
/**
* v1.18.10 P0 — fetch the per-night sleep summaries (with vitals) for the
* given window. `action=getsummary` keys by `lastupdate`/`startdateymd`; we
* use the same unix-second window the segment fetch uses, expressed as the
* `YYYY-MM-DD` form the summary endpoint expects. Returns the raw summaries
* (no DB write) so the caller composes the upsert step.
*/
export async function fetchWithingsSleepSummary(
accessToken: string,
startymd: string,
endymd: string,
): Promise<WithingsSleepSummary[]> {
const params = new URLSearchParams({
action: "getsummary",
startdateymd: startymd,
enddateymd: endymd,
data_fields: SLEEP_SUMMARY_DATA_FIELDS,
});
const pageStart = performance.now();
const res = await safeFetch(WITHINGS_SLEEP_URL, {
method: "POST",
headers: {
"Content-Type": "application/x-www-form-urlencoded",
Authorization: `Bearer ${accessToken}`,
},
body: params.toString(),
});
const json = await res.json();
const verdict = classifyWithingsResponse(res.status, json);
getEvent()?.addExternalCall({
service: "withings",
method: "fetchWithingsSleepSummary",
duration_ms: Math.round(performance.now() - pageStart),
status: res.status,
error: verdict.classification === "success" ? undefined : verdict.reason,
});
if (verdict.classification !== "success") {
throw new WithingsApiError({
verb: "sleep",
classification: verdict.classification,
withingsStatus: verdict.withingsStatus,
reason: verdict.reason,
upstreamError: typeof json?.error === "string" ? json.error : undefined,
});
}
const series: WithingsSleepSummary[] = json.body?.series ?? [];
return series;
}
/**
* Sync sleep data for a single user. Walks the trailing 30-day window
* (or whatever `SLEEP_BACKFILL_DAYS` is set to), writing one
@@ -319,6 +460,66 @@ export async function syncUserSleep(
segmentIndex++;
}
// v1.18.10 P0 — nightly sleep vitals (avg HR / respiratory rate / SDNN HRV
// / avg SpO2 / sleep score) from the per-night summary. A summary-fetch
// failure must not lose the stage segments already written, so it runs in
// its own try/catch and only annotates on error. Each vital upserts on the
// stable `(userId, type, source, externalId)` composite so a re-sync
// overwrites the night's value rather than inserting a duplicate.
try {
const startYmd = ymdUtc(start);
const endYmd = ymdUtc(now);
const summaries = await fetchWithingsSleepSummary(
tokenInfo.accessToken,
startYmd,
endYmd,
);
for (const summary of summaries) {
const measuredAt = new Date(summary.enddate * 1000);
const sessionId = summary.id ?? "no-id";
const vitals = mapWithingsSleepSummary(summary.data);
for (const vital of vitals) {
const externalId = `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`;
try {
await prisma.measurement.upsert({
where: {
userId_type_source_externalId: {
userId,
type: vital.type,
source: "WITHINGS",
externalId,
},
},
create: {
userId,
type: vital.type,
value: vital.value,
unit: getUnitForType(vital.type),
measuredAt,
source: "WITHINGS",
externalId,
},
update: {
value: vital.value,
measuredAt,
syncVersion: { increment: 1 },
},
});
touched.push({ type: vital.type, measuredAt });
imported++;
} catch (err) {
getEvent()?.addWarning(
`withings sleep: failed to upsert vital (${vital.type}, ${measuredAt.toISOString()}): ${err}`,
);
}
}
}
} catch (err) {
getEvent()?.addWarning(
`withings sleep: summary fetch failed for ${userId}: ${err}`,
);
}
// v1.4.39.1 — refresh the persistent rollup table for every distinct
// (type, day) the sync touched. Sleep segments collapse heavily —
// ~10 stage rows per night land in one DAY recompute.