Merge pull request #728 from MBombeck/fix/sync-range-validation

Sync writers refuse readings the app itself calls impossible
This commit is contained in:
Marc-André Bombeck authored and GitHub committed 2026-08-08 02:27:08 +02:00
commit abb0f2e82c
21 files changed
+1733 -8

No files matched your search

@@ -0,0 +1,150 @@
/**
* Read-only diagnostic for stored measurement rows outside the app's own
* declared plausibility domain.
*
* `VALUE_RANGES` in `@/lib/validations/measurement` is the band the
* application says a metric can occupy. Every path a person can drive has
* always enforced it; the provider sync writers did not, so rows the
* application itself calls impossible reached the table — and every mean,
* median and personal band built from them inherited the impossibility. The
* writers enforce it now, which stops new rows landing. It says nothing about
* the rows already there.
*
* This reports them and repairs nothing. What to do with a stored impossible
* row is a judgement the data cannot make: a pulse of 55 million is plainly a
* decode slip and a candidate for deletion, while a weight of 0.9 kg might be
* a neonatal record on a shared scale, or a unit mix-up, or a real reading the
* band is simply too narrow for. Deleting somebody's health record on a
* heuristic is not a thing a script gets to decide.
*
* Output is aggregate: one line per (metric, source) bucket with the row
* count, the number of distinct records affected, the observed extremes, and
* the window they fall in. Individual readings are not printed — the bucket
* plus the metric's own declared bound is what a decision needs, and a log
* file full of somebody's vitals is not.
*
* Run (inside the app container, which has DATABASE_URL):
* pnpm dlx tsx scripts/diagnose-implausible-measurements.ts
*/
import "dotenv/config";
import { prisma } from "@/lib/db";
import { VALUE_RANGES } from "@/lib/validations/measurement";
interface BucketRow {
type: string;
source: string;
row_count: bigint;
record_count: bigint;
min_value: number;
max_value: number;
first_at: Date;
last_at: Date;
tombstoned: bigint;
}
function ymd(date: Date): string {
return date.toISOString().slice(0, 10);
}
async function main(): Promise<void> {
const types = Object.keys(VALUE_RANGES).sort();
const mins = types.map((t) => VALUE_RANGES[t]!.min);
const maxes = types.map((t) => VALUE_RANGES[t]!.max);
// The band table rides in as three parallel bound arrays rather than a
// spliced VALUES list, so the query carries no interpolated text at all.
// `value <> value` is the NaN probe: a double-precision column can hold one,
// and it compares false against every bound including itself.
const buckets = await prisma.$queryRaw<BucketRow[]>`
WITH bands(type, min_value, max_value) AS (
SELECT * FROM unnest(
${types}::text[],
${mins}::double precision[],
${maxes}::double precision[]
)
)
SELECT
m."type"::text AS type,
m."source"::text AS source,
COUNT(*)::bigint AS row_count,
COUNT(DISTINCT m."user_id")::bigint AS record_count,
MIN(m."value") AS min_value,
MAX(m."value") AS max_value,
MIN(m."measured_at") AS first_at,
MAX(m."measured_at") AS last_at,
COUNT(*) FILTER (WHERE m."deleted_at" IS NOT NULL)::bigint AS tombstoned
FROM "measurements" m
JOIN bands b ON b."type" = m."type"::text
WHERE m."value" < b.min_value
OR m."value" > b.max_value
OR m."value" <> m."value"
GROUP BY m."type", m."source"
ORDER BY COUNT(*) DESC, m."type"::text ASC, m."source"::text ASC
`;
if (buckets.length === 0) {
console.log(
`Every stored measurement sits inside its metric's declared range ` +
`(${types.length} metric bands checked). Nothing to review.`,
);
return;
}
const totalRows = buckets.reduce((sum, b) => sum + Number(b.row_count), 0);
const affectedTypes = new Set(buckets.map((b) => b.type));
const affectedSources = new Set(buckets.map((b) => b.source));
console.log(
`${totalRows} stored row(s) sit outside the declared plausibility band, ` +
`across ${affectedTypes.size} metric(s) and ${affectedSources.size} source(s).\n` +
`Nothing below is modified. Each line is a (metric, source) bucket; the ` +
`band is the metric's own declared domain.\n`,
);
// Group by metric so the declared band is stated once per heading, and rank
// the metrics by how much of the problem each one is. The query's own
// ordering already ranks the sources inside a metric.
const byType = new Map<string, BucketRow[]>();
for (const bucket of buckets) {
const bucketsForType = byType.get(bucket.type);
if (bucketsForType) bucketsForType.push(bucket);
else byType.set(bucket.type, [bucket]);
}
const ranked = [...byType.entries()].sort(
(a, b) =>
b[1].reduce((sum, x) => sum + Number(x.row_count), 0) -
a[1].reduce((sum, x) => sum + Number(x.row_count), 0),
);
for (const [type, rows] of ranked) {
const band = VALUE_RANGES[type]!;
console.log(`${type} — declared band ${band.min}…${band.max}`);
for (const bucket of rows) {
const live = Number(bucket.row_count) - Number(bucket.tombstoned);
console.log(
` ${bucket.source.padEnd(14)} ` +
`${String(bucket.row_count).padStart(7)} row(s) ` +
`(${live} live, ${bucket.tombstoned} tombstoned) ` +
`across ${bucket.record_count} record(s) — ` +
`observed ${bucket.min_value}…${bucket.max_value}, ` +
`${ymd(bucket.first_at)} → ${ymd(bucket.last_at)}`,
);
}
}
console.log(
`\nThe sync writers refuse these values now, so the population above is ` +
`closed unless a writer without the gate is still in play. Removing a ` +
`row is a maintainer decision, per bucket.`,
);
}
main()
.catch((err) => {
console.error(err);
process.exitCode = 1;
})
.finally(() => {
void prisma.$disconnect();
});
@@ -0,0 +1,200 @@
/**
* Structural guard on the measurement ingest boundary.
*
* `VALUE_RANGES` is the band the application declares a metric can occupy.
* Every path a person can drive enforced it from the start; the provider sync
* writers never did, and nothing noticed, because each writer is correct on
* its own terms — it wrote what the remote sent. The gap was only visible from
* above, by listing every writer at once and asking which of them had ever
* been told about the band.
*
* So the list lives here. Every file that writes a `Measurement` row is
* enumerated with the reason it is allowed to, and the guard fails when the
* discovered set stops matching. A provider added tomorrow lands a file that
* is not in the map and has to answer the question before it can ship.
*
* This is a tripwire, not a proof. It cannot show a disposition is the right
* one — only that nobody added a writer without writing down which gate covers
* it. A reviewer who waves through a bogus `derived` label defeats it, and no
* test substitutes for that review.
*/
import { describe, it, expect } from "vitest";
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { walkSourceFiles } from "./helpers/source-files";
const SRC = join(process.cwd(), "src");
/** The gate module every provider-fed writer has to reach. */
const GATE_MODULE = "@/lib/measurements/plausibility-gate";
/** The predicate every person-driven write path has always used. */
const VALIDATOR = "validateMeasurementRange";
/**
* How a writer satisfies the plausibility domain.
*
* - `gate` — provider-fed; must reach {@link GATE_MODULE} itself.
* - `validator` — person-driven; must call {@link VALIDATOR} itself.
* - `upstream` — validated in a named module before the row is built; the
* named module is checked, so the label cannot be a shrug.
* - `derived` — the value is arithmetic over rows already in the table, not a
* number from outside. Gating here would drop a consolidation of data that
* IS present, which is a different failure from admitting one that is not.
* - `literal` — the written value is a constant in the file, so no external
* number can reach the column.
* - `restore` — replays the record's own backup verbatim. Fidelity is the
* contract; a restore that silently dropped rows would be the worse bug,
* and the diagnostic script is how a stored impossible row gets found.
*/
type Disposition =
| { kind: "gate" }
| { kind: "validator" }
| { kind: "upstream"; validatedIn: string }
| { kind: "derived"; why: string }
| { kind: "literal"; why: string }
| { kind: "restore"; why: string };
const WRITERS: Record<string, Disposition> = {
// ── Provider sync writers, gated ──────────────────────────────────────
"lib/withings/sync.ts": { kind: "gate" },
"lib/withings/sync-activity.ts": { kind: "gate" },
"lib/fitbit/sync-core.ts": { kind: "gate" },
"lib/google-health/sync-core.ts": { kind: "gate" },
"lib/nightscout/sync.ts": { kind: "gate" },
// The shared identity seam. Polar, Oura, WHOOP, the Withings sleep leg,
// the Apple export importer and the batch endpoint all write through it,
// so the gate here is what makes a new provider fail closed.
"lib/measurements/reconcile-external-measurement.ts": { kind: "gate" },
// ── Person-driven write paths, validated ──────────────────────────────
"app/api/measurements/route.ts": {
kind: "upstream",
validatedIn: "lib/validations/measurement.ts",
},
"app/api/import/route.ts": { kind: "validator" },
"lib/mcp/writes.ts": { kind: "validator" },
"lib/measurements/create-from-telegram.ts": { kind: "validator" },
"lib/measurements/import-apple-health-export.ts": { kind: "validator" },
"app/api/import/csv/route.ts": {
kind: "upstream",
validatedIn: "lib/import/csv-measurements.ts",
},
// ── Fixed values ──────────────────────────────────────────────────────
"lib/withings/sync-ecg.ts": {
kind: "literal",
why: "an ECG row carries value 1 — the file's own constant, never a provider number",
},
"app/api/mental-health/assessments/route.ts": {
kind: "literal",
why: "the questionnaire score is summed from Zod-bounded item answers",
},
"lib/insights/score-row.ts": {
kind: "literal",
why: "the health score is computed in-app on a closed 0-100 scale",
},
// ── Derived from rows already stored ──────────────────────────────────
"lib/measurements/consolidate-daily-mean.ts": {
kind: "derived",
why: "writes the mean of the day's own stored rows",
},
"lib/measurements/consolidate-legacy-steps.ts": {
kind: "derived",
why: "folds stored per-sample step rows into the day's canonical row",
},
"lib/measurements/dense-intraday-retention.ts": {
kind: "derived",
why: "rebuilds hourly buckets from stored per-sample rows",
},
"lib/measurements/drain-per-sample-cumulative.ts": {
kind: "derived",
why: "reduces stored cumulative per-sample rows to one daily total",
},
// ── Restore ───────────────────────────────────────────────────────────
"app/api/admin/backups/[id]/restore/route.ts": {
kind: "restore",
why: "replays the record's own export; dropping rows would break the restore contract",
},
};
/**
* Any Prisma client handle followed by a measurement write. Whitespace- and
* handle-tolerant on purpose: the earlier generation of this repo's guards
* demanded a literal `prisma.x.findUnique(` and went green because it matched
* nothing, not because nothing was there. The transaction handles in use are
* `tx` and `pc`; a future one is caught by the `[A-Za-z_$][\w$]*` head.
*/
const MEASUREMENT_WRITE =
/[A-Za-z_$][\w$]*\s*\.\s*measurement\s*\.\s*(create|createMany|createManyAndReturn|upsert)\s*\(/;
function sourceFiles(): string[] {
return walkSourceFiles(SRC, { floor: 3000 })
.filter((p) => !p.startsWith("generated/"))
.filter((p) => !p.includes("__tests__"))
.filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx"))
.sort();
}
function read(rel: string): string {
return readFileSync(join(SRC, rel), "utf8");
}
function discoveredWriters(): string[] {
return sourceFiles().filter((rel) => MEASUREMENT_WRITE.test(read(rel)));
}
describe("every measurement writer declares how it meets the plausibility domain", () => {
it("finds the writers at all — an empty sweep is a broken guard, not a clean repo", () => {
// The map is the claim; this asserts the matcher can still see the
// population it describes. A regex that silently stops matching would
// otherwise turn every check below into a tautology.
expect(discoveredWriters().length).toBeGreaterThanOrEqual(15);
});
it("the discovered writer set matches the declared map exactly", () => {
expect(discoveredWriters()).toEqual(Object.keys(WRITERS).sort());
});
it("every provider-fed writer reaches the gate module", () => {
const gated = Object.entries(WRITERS)
.filter(([, d]) => d.kind === "gate")
.map(([rel]) => rel);
expect(gated.length).toBeGreaterThanOrEqual(6);
for (const rel of gated) {
expect(read(rel), `${rel} must import ${GATE_MODULE}`).toContain(
GATE_MODULE,
);
}
});
it("every person-driven writer calls the range validator", () => {
for (const [rel, disposition] of Object.entries(WRITERS)) {
if (disposition.kind === "validator") {
expect(read(rel), `${rel} must call ${VALIDATOR}`).toContain(VALIDATOR);
}
if (disposition.kind === "upstream") {
expect(
read(disposition.validatedIn),
`${disposition.validatedIn} must call ${VALIDATOR} on behalf of ${rel}`,
).toContain(VALIDATOR);
}
}
});
it("the exempt writers carry a written reason", () => {
for (const [rel, disposition] of Object.entries(WRITERS)) {
if (
disposition.kind === "derived" ||
disposition.kind === "literal" ||
disposition.kind === "restore"
) {
expect(disposition.why.length, `${rel} needs a reason`).toBeGreaterThan(
20,
);
}
}
});
});
+11
View File
@@ -620,6 +620,17 @@ async function postBatch(request: NextRequest): Promise<Response> {
};
failedCount++;
continue;
case "rejected_range":
// The reconciler carries the same plausibility gate this route
// already applied above, so reaching here means the two disagreed —
// report it exactly as the earlier guard would have, and let the
// wide-event tally record which end refused.
results[p.index] = {
index: p.index,
status: "skipped",
reason: "value_out_of_range",
};
continue;
}
if (isMergeableSource(p.row.source as string)) {
@@ -0,0 +1,179 @@
/**
* `upsertFitbitMeasurements` is the single write seam for every Fitbit
* resource leg — activity, sleep, daily metrics — and it holds Prisma
* directly, so it does not inherit the shared reconciler's gate. These cases
* pin that the provider's numbers clear the app's own declared plausibility
* band before any of them can become a row, and that a refusal is reported.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const { findManyMock, createManyMock, updateMock } = vi.hoisted(() => ({
findManyMock: vi.fn(async () => [] as unknown[]),
createManyMock: vi.fn<
() => Promise<Array<{ id: string; type: string; measuredAt: Date }>>
>(async () => []),
updateMock: vi.fn(async () => ({})),
}));
vi.mock("@/lib/db", () => ({
prisma: {
measurement: {
findMany: findManyMock,
createManyAndReturn: createManyMock,
update: updateMock,
},
},
}));
vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() }));
vi.mock("@/lib/integrations/status", () => ({
isReauthRequired: vi.fn(async () => false),
recordSyncFailure: vi.fn(async () => {}),
recordSyncSuccess: vi.fn(async () => {}),
}));
vi.mock("@/lib/rollups/measurement-rollups", () => ({
collapseToTypeDayKeys: vi.fn(() => []),
recomputeBucketsForMeasurement: vi.fn(async () => {}),
recomputeUserRollups: vi.fn(async () => {}),
}));
vi.mock("@/lib/insights/comprehensive-generate", () => ({
invalidateStatusInsightsForTypes: vi.fn(async () => {}),
}));
vi.mock("@/lib/integrations/oauth-refresh", () => ({
persistRotatedToken: vi.fn(async () => {}),
}));
vi.mock("../credentials", () => ({
getUserFitbitCredentials: vi.fn(async () => null),
}));
vi.mock("../client", () => ({ refreshAccessToken: vi.fn() }));
import { eventStorage } from "@/lib/logging/context";
import { WideEventBuilder } from "@/lib/logging/event-builder";
import { RANGE_REJECTED_META_KEY } from "@/lib/measurements/plausibility-gate";
import type { RangeRejectionTally } from "@/lib/measurements/plausibility-gate";
import { upsertFitbitMeasurements } from "../sync-core";
/** RESTING_HEART_RATE is declared 25–220. */
const REAL_READING = {
type: "RESTING_HEART_RATE",
value: 58,
unit: "bpm",
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
externalId: "rhr:2026-07-08",
};
const IMPOSSIBLE_READING = {
type: "RESTING_HEART_RATE",
value: 4_915_200,
unit: "bpm",
measuredAt: new Date("2026-07-09T00:00:00.000Z"),
externalId: "rhr:2026-07-09",
};
let event: WideEventBuilder;
function tally(): RangeRejectionTally | undefined {
return event.toJSON().meta?.[RANGE_REJECTED_META_KEY] as
RangeRejectionTally | undefined;
}
beforeEach(() => {
findManyMock.mockReset().mockResolvedValue([]);
createManyMock.mockReset().mockResolvedValue([]);
updateMock.mockReset().mockResolvedValue({});
event = new WideEventBuilder("background");
});
describe("upsertFitbitMeasurements — plausibility gate", () => {
it("writes a reading inside the metric's declared band", async () => {
createManyMock.mockResolvedValue([
{
id: "row-1",
type: "RESTING_HEART_RATE",
measuredAt: REAL_READING.measuredAt,
},
]);
const result = await eventStorage.run(event, () =>
upsertFitbitMeasurements("user-1", [REAL_READING], { deferRollup: true }),
);
expect(result.imported).toBe(1);
const planned = (createManyMock.mock.calls[0] as unknown[])[0] as {
data: Array<{ value: number }>;
};
expect(planned.data).toHaveLength(1);
expect(planned.data[0]!.value).toBe(58);
expect(tally()).toBeUndefined();
});
it("refuses an out-of-band reading and annotates the drop", async () => {
createManyMock.mockResolvedValue([
{
id: "row-1",
type: "RESTING_HEART_RATE",
measuredAt: REAL_READING.measuredAt,
},
]);
const result = await eventStorage.run(event, () =>
upsertFitbitMeasurements("user-1", [REAL_READING, IMPOSSIBLE_READING], {
deferRollup: true,
}),
);
expect(result.imported).toBe(1);
const planned = (createManyMock.mock.calls[0] as unknown[])[0] as {
data: Array<{ value: number; externalId: string }>;
};
expect(planned.data).toHaveLength(1);
expect(planned.data[0]!.externalId).toBe("rhr:2026-07-08");
// The refused reading never even reached the existence probe.
expect(
(
(findManyMock.mock.calls[0] as unknown[])[0] as {
where: { externalId: { in: string[] } };
}
).where.externalId.in,
).toEqual(["rhr:2026-07-08"]);
expect(tally()).toEqual({
total: 1,
buckets: [
{
source: "fitbit",
type: "RESTING_HEART_RATE",
direction: "above_max",
count: 1,
},
],
});
expect(event.getLevel()).toBe("warn");
});
it("does not overwrite a stored reading with an out-of-band one", async () => {
// The key already exists; ungated, the update branch rewrote its value.
findManyMock.mockResolvedValue([
{
id: "existing",
type: "RESTING_HEART_RATE",
externalId: "rhr:2026-07-09",
value: 58,
unit: "bpm",
measuredAt: IMPOSSIBLE_READING.measuredAt,
sleepStage: null,
deletedAt: null,
},
]);
const result = await eventStorage.run(event, () =>
upsertFitbitMeasurements("user-1", [IMPOSSIBLE_READING], {
deferRollup: true,
}),
);
expect(result.imported).toBe(0);
expect(updateMock).not.toHaveBeenCalled();
expect(createManyMock).not.toHaveBeenCalled();
expect(tally()?.total).toBe(1);
});
});
+11 -1
View File
@@ -15,6 +15,7 @@ import { persistRotatedToken } from "@/lib/integrations/oauth-refresh";
import { recordSyncFailure } from "@/lib/integrations/status";
import type { FailureKind } from "@/lib/integrations/status";
import { getEvent } from "@/lib/logging/context";
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
import {
collapseToTypeDayKeys,
recomputeBucketsForMeasurement,
@@ -530,7 +531,7 @@ export async function replaceStaleFitbitSleep(
*/
export async function upsertFitbitMeasurements(
userId: string,
readings: FitbitMeasurementUpsert[],
incoming: FitbitMeasurementUpsert[],
opts: { deferRollup?: boolean } = {},
): Promise<{
imported: number;
@@ -541,6 +542,15 @@ export async function upsertFitbitMeasurements(
measuredAt: Date;
}>;
}> {
// Every resource leg funnels through here, so this is the one place the
// provider's numbers have to clear the app's declared plausibility band
// before they can become rows. A refused reading is dropped and tallied on
// the ambient event, never clamped and never written for a later reader to
// discover.
const readings = dropImplausibleMeasurements("fitbit", incoming, (r) => ({
type: r.type,
value: r.value,
}));
if (readings.length === 0) return { imported: 0, touched: [], inserted: [] };
// Probe EVERY existing row (live AND tombstoned) for the batch's externalIds
@@ -0,0 +1,180 @@
/**
* `upsertGoogleHealthMeasurements` is the single write seam for every Google
* Health resource leg and it holds Prisma directly, so it does not inherit the
* shared reconciler's gate. These cases pin that the provider's numbers clear
* the app's own declared plausibility band before any of them can become a
* row, and that a refusal is reported rather than swallowed.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const { findManyMock, createManyMock, updateMock } = vi.hoisted(() => ({
findManyMock: vi.fn(async () => [] as unknown[]),
createManyMock: vi.fn<
() => Promise<Array<{ id: string; type: string; measuredAt: Date }>>
>(async () => []),
updateMock: vi.fn(async () => ({})),
}));
vi.mock("@/lib/db", () => ({
prisma: {
measurement: {
findMany: findManyMock,
createManyAndReturn: createManyMock,
update: updateMock,
},
},
}));
vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() }));
vi.mock("@/lib/integrations/status", () => ({
isReauthRequired: vi.fn(async () => false),
recordSyncFailure: vi.fn(async () => {}),
recordSyncSuccess: vi.fn(async () => {}),
}));
vi.mock("@/lib/rollups/measurement-rollups", () => ({
collapseToTypeDayKeys: vi.fn(() => []),
recomputeBucketsForMeasurement: vi.fn(async () => {}),
recomputeUserRollups: vi.fn(async () => {}),
}));
vi.mock("@/lib/insights/comprehensive-generate", () => ({
invalidateStatusInsightsForTypes: vi.fn(async () => {}),
}));
vi.mock("../credentials", () => ({
getUserGoogleHealthCredentials: vi.fn(async () => null),
}));
vi.mock("../client", async () => {
const actual = await vi.importActual<typeof import("../client")>("../client");
return {
...actual,
refreshAccessToken: vi.fn(),
noteGoogleHealthMapped: vi.fn(),
noteGoogleHealthWritten: vi.fn(),
noteGoogleHealthOutcomeFailure: vi.fn(),
};
});
import { eventStorage } from "@/lib/logging/context";
import { WideEventBuilder } from "@/lib/logging/event-builder";
import { RANGE_REJECTED_META_KEY } from "@/lib/measurements/plausibility-gate";
import type { RangeRejectionTally } from "@/lib/measurements/plausibility-gate";
import { upsertGoogleHealthMeasurements } from "../sync-core";
/** OXYGEN_SATURATION is declared 50–100. */
const REAL_READING = {
type: "OXYGEN_SATURATION",
value: 97,
unit: "%",
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
externalId: "spo2:2026-07-08",
};
const IMPOSSIBLE_READING = {
type: "OXYGEN_SATURATION",
value: 9_700,
unit: "%",
measuredAt: new Date("2026-07-09T00:00:00.000Z"),
externalId: "spo2:2026-07-09",
};
let event: WideEventBuilder;
function tally(): RangeRejectionTally | undefined {
return event.toJSON().meta?.[RANGE_REJECTED_META_KEY] as
RangeRejectionTally | undefined;
}
beforeEach(() => {
findManyMock.mockReset().mockResolvedValue([]);
createManyMock.mockReset().mockResolvedValue([]);
updateMock.mockReset().mockResolvedValue({});
event = new WideEventBuilder("background");
});
describe("upsertGoogleHealthMeasurements — plausibility gate", () => {
it("writes a reading inside the metric's declared band", async () => {
createManyMock.mockResolvedValue([
{
id: "row-1",
type: "OXYGEN_SATURATION",
measuredAt: REAL_READING.measuredAt,
},
]);
const result = await eventStorage.run(event, () =>
upsertGoogleHealthMeasurements("user-1", [REAL_READING], {
deferRollup: true,
}),
);
expect(result.imported).toBe(1);
const planned = (createManyMock.mock.calls[0] as unknown[])[0] as {
data: Array<{ value: number }>;
};
expect(planned.data).toHaveLength(1);
expect(planned.data[0]!.value).toBe(97);
expect(tally()).toBeUndefined();
});
it("refuses an out-of-band reading and annotates the drop", async () => {
createManyMock.mockResolvedValue([
{
id: "row-1",
type: "OXYGEN_SATURATION",
measuredAt: REAL_READING.measuredAt,
},
]);
const result = await eventStorage.run(event, () =>
upsertGoogleHealthMeasurements(
"user-1",
[REAL_READING, IMPOSSIBLE_READING],
{ deferRollup: true },
),
);
expect(result.imported).toBe(1);
const planned = (createManyMock.mock.calls[0] as unknown[])[0] as {
data: Array<{ externalId: string }>;
};
expect(planned.data).toHaveLength(1);
expect(planned.data[0]!.externalId).toBe("spo2:2026-07-08");
expect(tally()).toEqual({
total: 1,
buckets: [
{
source: "googleHealth",
type: "OXYGEN_SATURATION",
direction: "above_max",
count: 1,
},
],
});
expect(event.getLevel()).toBe("warn");
});
it("does not overwrite a stored reading with an out-of-band one", async () => {
findManyMock.mockResolvedValue([
{
id: "existing",
type: "OXYGEN_SATURATION",
externalId: "spo2:2026-07-09",
value: 97,
unit: "%",
measuredAt: IMPOSSIBLE_READING.measuredAt,
sleepStage: null,
deletedAt: null,
},
]);
const result = await eventStorage.run(event, () =>
upsertGoogleHealthMeasurements("user-1", [IMPOSSIBLE_READING], {
deferRollup: true,
}),
);
expect(result.imported).toBe(0);
expect(updateMock).not.toHaveBeenCalled();
expect(createManyMock).not.toHaveBeenCalled();
expect(tally()?.total).toBe(1);
});
});
+13 -2
View File
@@ -20,6 +20,7 @@ import { prisma } from "@/lib/db";
import type { MeasurementType } from "@/generated/prisma/client";
import { encrypt, decrypt } from "@/lib/crypto";
import { getEvent } from "@/lib/logging/context";
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status";
import {
@@ -487,7 +488,7 @@ export async function replaceStaleGoogleHealthSleep(
*/
export async function upsertGoogleHealthMeasurements(
userId: string,
readings: GoogleHealthMeasurementUpsert[],
incoming: GoogleHealthMeasurementUpsert[],
opts: { deferRollup?: boolean } = {},
): Promise<{
imported: number;
@@ -498,7 +499,17 @@ export async function upsertGoogleHealthMeasurements(
measuredAt: Date;
}>;
}> {
noteGoogleHealthMapped(readings.length);
noteGoogleHealthMapped(incoming.length);
// Every resource leg funnels through here, so this is the one place the
// provider's numbers have to clear the app's declared plausibility band
// before they can become rows. A refused reading is dropped and tallied on
// the ambient event, never clamped and never written for a later reader to
// discover.
const readings = dropImplausibleMeasurements(
"googleHealth",
incoming,
(r) => ({ type: r.type, value: r.value }),
);
if (readings.length === 0) return { imported: 0, touched: [], inserted: [] };
// Probe EVERY existing row (live AND tombstoned) for the batch's externalIds
@@ -0,0 +1,222 @@
/**
* The ingest-side plausibility gate.
*
* `VALUE_RANGES` is the band the application declares a metric can occupy, and
* every path a person can drive has always enforced it. The provider sync
* writers did not, which is how a row the application itself calls impossible
* reached the table and poisoned every mean and personal band derived from it.
* These cases pin the two halves of the fix: the refusal, and the fact that a
* refusal is never silent.
*/
import { beforeEach, describe, expect, it } from "vitest";
import { WideEventBuilder } from "@/lib/logging/event-builder";
import { eventStorage } from "@/lib/logging/context";
import {
RANGE_REJECTED_META_KEY,
classifyRangeRejection,
dropImplausibleMeasurements,
recordRangeRejections,
type RangeRejectionTally,
} from "../plausibility-gate";
let event: WideEventBuilder;
function withEvent<T>(run: () => T): T {
return eventStorage.run(event, run);
}
function tally(): RangeRejectionTally | undefined {
return event.toJSON().meta?.[RANGE_REJECTED_META_KEY] as
RangeRejectionTally | undefined;
}
beforeEach(() => {
event = new WideEventBuilder("background");
});
describe("classifyRangeRejection", () => {
it("passes a value inside the metric's declared band", () => {
// PULSE is declared 20–300; both edges are inclusive.
expect(classifyRangeRejection("PULSE", 60)).toBeNull();
expect(classifyRangeRejection("PULSE", 20)).toBeNull();
expect(classifyRangeRejection("PULSE", 300)).toBeNull();
});
it("names the edge an out-of-band value broke", () => {
expect(classifyRangeRejection("PULSE", 55643821.505)).toEqual({
type: "PULSE",
direction: "above_max",
});
expect(classifyRangeRejection("PULSE", 19)).toEqual({
type: "PULSE",
direction: "below_min",
});
});
it("refuses a non-finite value whatever the metric", () => {
expect(classifyRangeRejection("PULSE", Number.NaN)).toEqual({
type: "PULSE",
direction: "not_finite",
});
// No declared band, so the value itself is the only thing to judge.
expect(
classifyRangeRejection("NOT_A_METRIC", Number.POSITIVE_INFINITY),
).toEqual({ type: "NOT_A_METRIC", direction: "not_finite" });
});
it("passes a metric with no declared band — an absent range is an absent fact", () => {
expect(classifyRangeRejection("NOT_A_METRIC", 9_999_999)).toBeNull();
});
});
describe("dropImplausibleMeasurements", () => {
it("keeps the in-band rows and returns them untouched", () => {
const rows = [
{ type: "PULSE", value: 60 },
{ type: "WEIGHT", value: 82.4 },
];
const kept = withEvent(() =>
dropImplausibleMeasurements("withings", rows, (r) => r),
);
expect(kept).toEqual(rows);
expect(kept[0]).toBe(rows[0]);
// Nothing was refused, so the event carries no tally at all — absence
// reads as absence.
expect(tally()).toBeUndefined();
expect(event.toJSON().warnings).toBeUndefined();
});
it("drops the out-of-band rows and tallies them on the wide event", () => {
const kept = withEvent(() =>
dropImplausibleMeasurements(
"withings",
[
{ type: "PULSE", value: 60 },
{ type: "PULSE", value: 55643821.505 },
{ type: "PULSE", value: 900 },
{ type: "WEIGHT", value: 0.2 },
],
(r) => r,
),
);
expect(kept).toEqual([{ type: "PULSE", value: 60 }]);
const recorded = tally();
expect(recorded).toEqual({
total: 3,
buckets: [
{
source: "withings",
type: "PULSE",
direction: "above_max",
count: 2,
},
{
source: "withings",
type: "WEIGHT",
direction: "below_min",
count: 1,
},
],
});
});
it("keeps the refused readings out of the log line", () => {
withEvent(() =>
dropImplausibleMeasurements(
"withings",
[{ type: "PULSE", value: 55643821.505 }],
(r) => r,
),
);
// The value is a health reading. The tally says which metric, which
// provider and which edge — never the number itself.
const serialised = JSON.stringify(event.toJSON());
expect(serialised).not.toContain("55643821");
});
it("lifts the event to warn once per provider", () => {
withEvent(() => {
dropImplausibleMeasurements(
"withings",
[{ type: "PULSE", value: 900 }],
(r) => r,
);
dropImplausibleMeasurements(
"withings",
[{ type: "PULSE", value: 901 }],
(r) => r,
);
dropImplausibleMeasurements(
"fitbit",
[{ type: "PULSE", value: 902 }],
(r) => r,
);
});
expect(event.getLevel()).toBe("warn");
expect(event.toJSON().warnings).toHaveLength(2);
expect(event.toJSON().warnings?.[0]).toContain("withings");
expect(event.toJSON().warnings?.[1]).toContain("fitbit");
});
it("accumulates across calls instead of overwriting the previous tally", () => {
// Fitbit gates once per resource leg, and the reconciler gates once per
// row: a tally that replaced rather than merged would report only the
// last batch, and a run that dropped hundreds would read as dropping one.
withEvent(() => {
dropImplausibleMeasurements(
"fitbit",
[{ type: "PULSE", value: 900 }],
(r) => r,
);
dropImplausibleMeasurements(
"fitbit",
[
{ type: "PULSE", value: 901 },
{ type: "OXYGEN_SATURATION", value: 4 },
],
(r) => r,
);
});
expect(tally()).toEqual({
total: 3,
buckets: [
{
source: "fitbit",
type: "OXYGEN_SATURATION",
direction: "below_min",
count: 1,
},
{ source: "fitbit", type: "PULSE", direction: "above_max", count: 2 },
],
});
});
it("still drops outside a wide-event context", () => {
// A sync running without an ambient event has nowhere to report. The row
// is refused anyway: observability is what is lost, never the guarantee.
const kept = dropImplausibleMeasurements(
"nightscout",
[
{ type: "BLOOD_GLUCOSE", value: 110 },
{ type: "BLOOD_GLUCOSE", value: 9_000 },
],
(r) => r,
);
expect(kept).toEqual([{ type: "BLOOD_GLUCOSE", value: 110 }]);
});
});
describe("recordRangeRejections", () => {
it("writes nothing when there is nothing to report", () => {
withEvent(() => recordRangeRejections("withings", []));
expect(tally()).toBeUndefined();
expect(event.getLevel()).toBe("info");
});
});
@@ -1,6 +1,9 @@
import { describe, expect, it, vi } from "vitest";
import type { Prisma } from "@/generated/prisma/client";
import { eventStorage } from "@/lib/logging/context";
import { WideEventBuilder } from "@/lib/logging/event-builder";
import { RANGE_REJECTED_META_KEY } from "../plausibility-gate";
import {
reconcileExternalMeasurement,
type ExternalMeasurementWrite,
@@ -383,3 +386,104 @@ describe("reconcileExternalMeasurement", () => {
);
});
});
/**
* The reconciler is the shared write seam for every provider that does not
* hold Prisma itself — Polar, Oura, WHOOP, the Withings sleep leg, the Apple
* export importer, the batch endpoint. Gating here rather than at each caller
* is what makes a provider added tomorrow inherit the refusal instead of
* having to remember a check it never knew about.
*/
describe("reconcileExternalMeasurement — plausibility gate", () => {
it("reconciles a value inside the metric's declared band", async () => {
const { tx, measurement } = transaction([]);
measurement.create.mockResolvedValue(row({ id: "fresh" }));
// PULSE is declared 20–300.
const verdict = await reconcileExternalMeasurement(
tx,
desired({ type: "PULSE", value: 112, unit: "bpm", sleepStage: null }),
);
expect(verdict.status).toBe("inserted");
expect(measurement.create).toHaveBeenCalledTimes(1);
});
it("refuses an out-of-band value without touching the transaction", async () => {
const { tx, measurement } = transaction([]);
const verdict = await reconcileExternalMeasurement(
tx,
desired({
type: "PULSE",
value: 55643821.505,
unit: "bpm",
sleepStage: null,
}),
);
expect(verdict).toEqual({
status: "rejected_range",
rejection: { type: "PULSE", direction: "above_max" },
});
// No row written, no row updated, and no savepoint opened — the refusal
// happens before the reconciler does any work at all.
expect(measurement.create).not.toHaveBeenCalled();
expect(measurement.update).not.toHaveBeenCalled();
expect(tx.$executeRawUnsafe).not.toHaveBeenCalled();
});
it("refuses rather than overwriting a good stored row", async () => {
// The dangerous shape: a live row already holds a real reading, and the
// provider re-sends the same identity carrying an impossible value. The
// update branch would have rewritten a good value with a bad one.
const existing = row({
id: "exact-row",
type: "PULSE",
value: 62,
unit: "bpm",
sleepStage: null,
measuredAt: new Date("2026-07-20T06:30:00.000Z"),
});
const { tx, measurement } = transaction([existing]);
const verdict = await reconcileExternalMeasurement(
tx,
desired({
type: "PULSE",
value: 55643821.505,
unit: "bpm",
sleepStage: null,
}),
);
expect(verdict.status).toBe("rejected_range");
expect(measurement.update).not.toHaveBeenCalled();
});
it("tallies the refusal on the ambient wide event under the provider", async () => {
const { tx } = transaction([]);
const event = new WideEventBuilder("background");
await eventStorage.run(event, () =>
reconcileExternalMeasurement(
tx,
desired({
type: "PULSE",
value: 55643821.505,
unit: "bpm",
source: "WHOOP",
sleepStage: null,
}),
),
);
expect(event.toJSON().meta?.[RANGE_REJECTED_META_KEY]).toEqual({
total: 1,
buckets: [
{ source: "whoop", type: "PULSE", direction: "above_max", count: 1 },
],
});
expect(event.getLevel()).toBe("warn");
});
});
@@ -720,6 +720,16 @@ export async function streamParseExportXml(
),
);
// The reconciler carries the same plausibility gate the block above
// already applied, so this arm means the two disagreed. Tally it under
// the same counter the earlier guard uses and count no row.
if (verdict.status === "rejected_range") {
unknown[`${type}::aggregate_out_of_range`] =
(unknown[`${type}::aggregate_out_of_range`] ?? 0) + 1;
stat.durationMs += Date.now() - rowStart;
continue;
}
cumulativeEstimatedDays.add(dayKey);
cumulativeEstimatedRows += 1;
if (verdict.status === "inserted") {
+192
View File
@@ -0,0 +1,192 @@
/**
* The ingest-side half of the app's plausibility domain.
*
* `VALUE_RANGES` in `@/lib/validations/measurement` is the band the
* application declares a metric can occupy — PULSE 20–300, WEIGHT 1–500,
* SLEEP_DURATION 0–1440 minutes. Every path a person can drive enforces it:
* the measurement routes, the batch endpoint, the CSV and Apple-export
* importers, MCP, Telegram. The provider sync writers did not. They took the
* number the remote handed over and wrote it, so a decode slip or a glitched
* device landed a row the application itself calls impossible, and every mean,
* median and personal band downstream digested it as a reading.
*
* `isPlausibleMetricValue` (in `value-domain.ts`) already answers the question
* for the READ side, which is where the damage surfaced first. This module is
* the same question asked one layer earlier, at the point of writing, so the
* row never lands. The two ends share the one predicate on purpose: a band the
* reader hides but the writer accepts is a row that exists and cannot be seen,
* which is the worse of the two failures.
*
* ## Disposition: drop, and say so
*
* An out-of-band incoming value is DROPPED. Not clamped (a clamp invents a
* reading nobody took), not stored-and-flagged (a quarantined row is still a
* row, and every consumer would need to learn the flag), not fatal to the
* batch (one glitched sample must not strand the sync's other readings behind
* a held watermark). Absence reads as absence: the sample is simply not part
* of the record.
*
* A silent drop, though, is indistinguishable from a provider that sent
* nothing — so every drop is tallied onto the ambient wide event under
* {@link RANGE_REJECTED_META_KEY}, with the metric type, the provider, and the
* direction it broke the band in. Counts only. The value itself is a health
* reading and never enters the log line; the direction plus the metric's own
* declared bound is what a diagnosis actually needs.
*
* ## Why meta and not `action`
*
* A wide event holds exactly ONE action, and `setAction` replaces it. A drop
* happens in the middle of a sync whose action is decided at the end, so an
* `action` set here is either clobbered a moment later (the drop vanishes) or,
* on the paths that annotate no terminal action, relabels an otherwise
* successful sync as a rejection. Meta accumulates, survives every later
* `setAction`, and is the shape a dashboard can sum. One warning per provider
* per event rides along so the drop also lifts the event's level to `warn`
* rather than hiding inside a green line.
*/
import { annotate, getEvent } from "@/lib/logging/context";
import { isPlausibleMetricValue } from "@/lib/measurements/value-domain";
import { VALUE_RANGES } from "@/lib/validations/measurement";
/**
* Stable meta key the tally lands under. Pinned so dashboards can count it
* across providers without knowing which one wrote the event.
*/
export const RANGE_REJECTED_META_KEY = "measurement.range_rejected";
/** Which edge of the declared band the value broke, or that it was not a number. */
export type RangeRejectionDirection = "below_min" | "above_max" | "not_finite";
/** One refused reading, reduced to the facts that carry no health value. */
export interface MeasurementRangeRejection {
/** `MeasurementType` of the refused reading. */
type: string;
direction: RangeRejectionDirection;
}
/** One (provider, metric, direction) cell of the event's running tally. */
export interface RangeRejectionBucket {
source: string;
type: string;
direction: RangeRejectionDirection;
count: number;
}
/** The pinned meta shape written under {@link RANGE_REJECTED_META_KEY}. */
export interface RangeRejectionTally {
total: number;
buckets: RangeRejectionBucket[];
}
/**
* Classify a candidate write against the metric's declared band.
*
* Returns `null` when the value may be written. A type with no declared range
* passes — an absent range is an absent fact, never a guessed one, and the
* same asymmetry the read side uses (`isPlausibleMetricValue`). A non-finite
* value never passes, whatever the type: NaN is not an input to anything.
*/
export function classifyRangeRejection(
type: string,
value: number,
): MeasurementRangeRejection | null {
if (isPlausibleMetricValue(type, value)) return null;
if (!Number.isFinite(value)) return { type, direction: "not_finite" };
const range = VALUE_RANGES[type];
// `isPlausibleMetricValue` only answers false for a finite value when the
// type HAS a range, so the lookup cannot miss here.
return {
type,
direction: value < range!.min ? "below_min" : "above_max",
};
}
function readTally(): RangeRejectionTally {
const existing = getEvent()?.toJSON().meta?.[RANGE_REJECTED_META_KEY];
if (
existing !== null &&
typeof existing === "object" &&
Array.isArray((existing as RangeRejectionTally).buckets)
) {
return existing as RangeRejectionTally;
}
return { total: 0, buckets: [] };
}
/**
* Fold refusals onto the ambient wide event.
*
* Merges with whatever the event already carries, so a provider that gates
* several batches (or reconciles row by row) accumulates instead of
* overwriting. Outside a wide-event context this is a no-op — the caller still
* drops the row, it simply has nowhere to report.
*/
export function recordRangeRejections(
source: string,
rejections: readonly MeasurementRangeRejection[],
): void {
if (rejections.length === 0) return;
const evt = getEvent();
if (!evt) return;
const tally = readTally();
const firstForSource = !tally.buckets.some((b) => b.source === source);
for (const rejection of rejections) {
const bucket = tally.buckets.find(
(b) =>
b.source === source &&
b.type === rejection.type &&
b.direction === rejection.direction,
);
if (bucket) bucket.count += 1;
else
tally.buckets.push({
source,
type: rejection.type,
direction: rejection.direction,
count: 1,
});
tally.total += 1;
}
tally.buckets.sort(
(a, b) =>
a.source.localeCompare(b.source) ||
a.type.localeCompare(b.type) ||
a.direction.localeCompare(b.direction),
);
annotate({ meta: { [RANGE_REJECTED_META_KEY]: tally } });
if (firstForSource) {
evt.addWarning(
`${source}: refused reading(s) outside the declared plausibility range — ` +
`dropped, see meta.${RANGE_REJECTED_META_KEY}`,
);
}
}
/**
* Keep the rows whose value sits inside the metric's declared band; tally and
* drop the rest.
*
* `select` exists because the sync writers carry their own row shapes and none
* of them is worth reshaping at the call site: the gate reads the two fields
* it needs and returns the caller's own objects untouched.
*/
export function dropImplausibleMeasurements<T>(
source: string,
rows: readonly T[],
select: (row: T) => { type: string; value: number },
): T[] {
const kept: T[] = [];
const rejections: MeasurementRangeRejection[] = [];
for (const row of rows) {
const { type, value } = select(row);
const rejection = classifyRangeRejection(type, value);
if (rejection) rejections.push(rejection);
else kept.push(row);
}
recordRangeRejections(source, rejections);
return kept;
}
@@ -4,6 +4,11 @@ import {
type MeasurementAggregationProvenance,
type Prisma as PrismaTypes,
} from "@/generated/prisma/client";
import {
classifyRangeRejection,
recordRangeRejections,
type MeasurementRangeRejection,
} from "@/lib/measurements/plausibility-gate";
const SAVEPOINT = "measurement_identity_reconcile";
@@ -50,6 +55,18 @@ export type MeasurementReconciliationVerdict =
| {
status: "failed";
error: { message: string; code?: string };
}
| {
/**
* The incoming value sits outside the metric's declared plausibility
* band, so nothing was written and no existing row was touched. Not a
* failure: the write was refused on purpose and the caller must carry
* on with the rest of its batch. Every caller narrows on this status
* because it carries no `row` — the compiler is what makes a future
* provider handle it rather than dereference a row that does not exist.
*/
status: "rejected_range";
rejection: MeasurementRangeRejection;
};
export type FailedMeasurementReconciliation = Extract<
@@ -219,12 +236,26 @@ async function retireCollision(
* advisory lock serializes cooperative writers sharing either identity, and a
* savepoint turns a non-benign database error into a structured `failed`
* verdict without poisoning the caller's surrounding transaction.
*
* This is also the plausibility gate for every provider that reconciles rather
* than writing Prisma directly. Gating HERE rather than at each caller is the
* point: a provider added tomorrow inherits the refusal by construction, and
* cannot land an impossible reading by forgetting a check it never knew about.
*/
export async function reconcileExternalMeasurement(
tx: PrismaTypes.TransactionClient,
desired: ExternalMeasurementWrite,
options: MeasurementReconciliationOptions = {},
): Promise<MeasurementReconciliationVerdict> {
const rejection = classifyRangeRejection(desired.type, desired.value);
if (rejection) {
recordRangeRejections(
desired.source ? String(desired.source).toLowerCase() : "external",
[rejection],
);
return { status: "rejected_range", rejection };
}
await tx.$executeRawUnsafe(`SAVEPOINT ${SAVEPOINT}`);
try {
@@ -0,0 +1,115 @@
/**
* The Nightscout pull writes BLOOD_GLUCOSE rows directly, so it does not
* inherit the shared reconciler's gate. It is also first-write-wins: a row
* that lands here is immutable, so an impossible sample never corrects itself
* on a later pass. These cases pin the refusal and its annotation.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const {
upsertMock,
findManyMock,
emitArrivalsMock,
recomputeMock,
invalidateMock,
} = vi.hoisted(() => ({
upsertMock: vi.fn(),
findManyMock: vi.fn(),
emitArrivalsMock: vi.fn(),
recomputeMock: vi.fn(),
invalidateMock: vi.fn(),
}));
vi.mock("@/lib/db", () => ({
prisma: { measurement: { upsert: upsertMock, findMany: findManyMock } },
}));
vi.mock("@/lib/arrivals/measurement-emit", () => ({
emitInsertedMeasurementArrivals: emitArrivalsMock,
}));
vi.mock("@/lib/rollups/measurement-rollups", () => ({
collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) =>
rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })),
recomputeBucketsForMeasurement: recomputeMock,
}));
vi.mock("@/lib/insights/comprehensive-generate", () => ({
invalidateStatusInsightsForTypes: invalidateMock,
}));
import { eventStorage } from "@/lib/logging/context";
import { WideEventBuilder } from "@/lib/logging/event-builder";
import { RANGE_REJECTED_META_KEY } from "@/lib/measurements/plausibility-gate";
import type { RangeRejectionTally } from "@/lib/measurements/plausibility-gate";
import { upsertNightscoutEntries } from "../sync";
/** BLOOD_GLUCOSE is declared 20–800 mg/dL. */
const REAL_ENTRY = { id: "abc", sgv: 112, date: 1718000000000 };
/** A sensor fault, or an mg/dL field carrying something else entirely. */
const IMPOSSIBLE_ENTRY = { id: "def", sgv: 41_600, date: 1718000300000 };
let event: WideEventBuilder;
function tally(): RangeRejectionTally | undefined {
return event.toJSON().meta?.[RANGE_REJECTED_META_KEY] as
RangeRejectionTally | undefined;
}
beforeEach(() => {
vi.clearAllMocks();
findManyMock.mockResolvedValue([]);
upsertMock.mockResolvedValue({ id: "row-1" });
emitArrivalsMock.mockResolvedValue(undefined);
recomputeMock.mockResolvedValue(undefined);
invalidateMock.mockResolvedValue(undefined);
event = new WideEventBuilder("background");
});
describe("upsertNightscoutEntries — plausibility gate", () => {
it("writes a reading inside the metric's declared band", async () => {
const result = await eventStorage.run(event, () =>
upsertNightscoutEntries("user-1", [REAL_ENTRY]),
);
expect(result).toEqual({ inserted: 1, failedRows: 0 });
expect(upsertMock).toHaveBeenCalledTimes(1);
expect(tally()).toBeUndefined();
});
it("refuses an out-of-band reading and annotates the drop", async () => {
const result = await eventStorage.run(event, () =>
upsertNightscoutEntries("user-1", [REAL_ENTRY, IMPOSSIBLE_ENTRY]),
);
expect(result).toEqual({ inserted: 1, failedRows: 0 });
expect(upsertMock).toHaveBeenCalledTimes(1);
expect(
(upsertMock.mock.calls[0]![0] as { create: { value: number } }).create
.value,
).toBe(112);
expect(tally()).toEqual({
total: 1,
buckets: [
{
source: "nightscout",
type: "BLOOD_GLUCOSE",
direction: "above_max",
count: 1,
},
],
});
expect(event.getLevel()).toBe("warn");
});
it("writes nothing when the whole page is out of band", async () => {
const result = await eventStorage.run(event, () =>
upsertNightscoutEntries("user-1", [IMPOSSIBLE_ENTRY]),
);
expect(result).toEqual({ inserted: 0, failedRows: 0 });
expect(upsertMock).not.toHaveBeenCalled();
expect(tally()?.total).toBe(1);
});
});
+11 -1
View File
@@ -22,6 +22,7 @@
import { prisma } from "@/lib/db";
import type { MeasurementType } from "@/generated/prisma/client";
import { getEvent } from "@/lib/logging/context";
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
import {
markSyncFailureRecorded,
recordSyncFailure,
@@ -221,7 +222,16 @@ export async function upsertNightscoutEntries(
if (entries.length === 0) return { inserted: 0, failedRows: 0 };
const type = "BLOOD_GLUCOSE" as MeasurementType;
const mapped = entries.map((entry) => mapSgvEntryToMeasurement(entry));
// A CGM that reports a glucose the app's own band calls impossible is
// reporting a sensor fault or a unit slip, not a reading. Drop it: the row
// would otherwise be immutable (first-write-wins), so an impossible sample
// that lands here never corrects itself on a later pass.
const mapped = dropImplausibleMeasurements(
"nightscout",
entries.map((entry) => mapSgvEntryToMeasurement(entry)),
(m) => ({ type, value: m.value }),
);
if (mapped.length === 0) return { inserted: 0, failedRows: 0 };
// Probe which keys already exist so a re-confirmed reading (the upsert's
// no-op update branch) is not miscounted as a fresh import and the arrival
+4
View File
@@ -575,6 +575,10 @@ export async function upsertOuraMeasurements(
for (let index = 0; index < readings.length; index++) {
const reading = readings[index]!;
const verdict = verdicts[index]!;
// Refused by the plausibility gate: nothing was written and nothing was
// touched, so the reading counts as neither imported nor a rollup dirt
// mark. The drop is already on the wide event.
if (verdict.status === "rejected_range") continue;
for (const dirty of verdict.dirtyIdentities ?? []) {
touched.push(dirty);
}
+4
View File
@@ -364,6 +364,10 @@ export async function upsertPolarMeasurements(
for (let index = 0; index < readings.length; index++) {
const reading = readings[index]!;
const verdict = verdicts[index]!;
// Refused by the plausibility gate: nothing was written and nothing was
// touched, so the reading counts as neither imported nor a rollup dirt
// mark. The drop is already on the wide event.
if (verdict.status === "rejected_range") continue;
for (const dirty of verdict.dirtyIdentities ?? []) {
touched.push(dirty);
}
+4
View File
@@ -480,6 +480,10 @@ export async function upsertWhoopMeasurements(
for (let index = chunkStart; index < chunkEnd; index++) {
const reading = readings[index]!;
const verdict = verdicts[index - chunkStart]!;
// Refused by the plausibility gate: nothing was written and nothing was
// touched, so the reading counts as neither imported nor a rollup dirt
// mark. The drop is already on the wide event.
if (verdict.status === "rejected_range") continue;
for (const dirty of verdict.dirtyIdentities ?? []) {
touched.push(dirty);
touchedTypes.add(dirty.type);
@@ -0,0 +1,261 @@
/**
* Withings writes the measurement table directly — the measures leg row by
* row, the activity leg as a planned batch — so neither inherits the shared
* reconciler's gate. These cases pin that both legs now clear the app's own
* declared plausibility band before anything reaches Prisma, and that a
* refusal lands on the wide event rather than vanishing.
*
* The measures leg matters twice over: it does not only insert, it OVERWRITES
* the stored value of an existing slot, so an ungated impossible reading could
* replace a real one.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
vi.mock("@/lib/db", () => ({
prisma: {
measurement: {
findFirst: vi.fn(),
findMany: vi.fn(),
create: vi.fn(),
createMany: vi.fn(),
update: vi.fn(),
},
withingsConnection: {
findUnique: vi.fn(),
update: vi.fn(),
},
$queryRaw: vi.fn(),
$transaction: vi.fn((ops: unknown) =>
Array.isArray(ops) ? Promise.all(ops) : Promise.resolve(),
),
},
}));
vi.mock("@/lib/crypto", () => ({
decrypt: vi.fn((v: string) => v),
encrypt: vi.fn((v: string) => v),
}));
vi.mock("../client", async () => {
const actual = await vi.importActual<typeof import("../client")>("../client");
return {
...actual,
fetchMeasurements: vi.fn(),
refreshAccessToken: vi.fn(),
subscribeWebhook: vi.fn(),
};
});
vi.mock("../credentials", () => ({
getUserWithingsCredentials: vi.fn(async () => ({
clientId: "cid",
clientSecret: "secret",
})),
}));
vi.mock("@/lib/integrations/status", () => ({
isReauthRequired: vi.fn().mockResolvedValue(false),
parkIntegrationAtReauth: vi.fn(),
recordSyncFailure: vi.fn().mockResolvedValue(undefined),
recordSyncSuccess: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/rollups/measurement-rollups", async () => {
const actual = await vi.importActual<
typeof import("@/lib/rollups/measurement-rollups")
>("@/lib/rollups/measurement-rollups");
return {
...actual,
recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
};
});
vi.mock("@/lib/insights/comprehensive-generate", () => ({
invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/cache/invalidate", () => ({
invalidateUserDashboardSnapshot: vi.fn(),
}));
import { prisma } from "@/lib/db";
import { eventStorage } from "@/lib/logging/context";
import { WideEventBuilder } from "@/lib/logging/event-builder";
import { RANGE_REJECTED_META_KEY } from "@/lib/measurements/plausibility-gate";
import type { RangeRejectionTally } from "@/lib/measurements/plausibility-gate";
import { fetchMeasurements } from "../client";
import { syncUserActivity } from "../sync-activity";
import { syncUserMeasurements } from "../sync";
let event: WideEventBuilder;
function tally(): RangeRejectionTally | undefined {
return event.toJSON().meta?.[RANGE_REJECTED_META_KEY] as
RangeRejectionTally | undefined;
}
function installActivityFetch(
entries: Array<{ date: string; steps?: number; distance?: number }>,
) {
vi.stubGlobal(
"fetch",
vi.fn(async () => ({
status: 200,
headers: new Headers(),
json: async () => ({
status: 0,
body: { activities: entries, more: false, offset: 0 },
}),
})),
);
}
beforeEach(() => {
vi.clearAllMocks();
event = new WideEventBuilder("background");
vi.mocked(prisma.withingsConnection.findUnique).mockResolvedValue({
id: "conn-1",
withingsUserId: "wu-1",
accessToken: "enc-access",
refreshToken: "enc-refresh",
scope: "user.metrics,user.activity",
tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000),
} as never);
vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never);
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
vi.mocked(prisma.measurement.create).mockResolvedValue({
id: "m-1",
type: "PULSE",
measuredAt: new Date(),
} as never);
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
count: 1,
} as never);
vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
});
afterEach(() => {
vi.unstubAllGlobals();
});
describe("syncUserMeasurements — plausibility gate", () => {
it("writes a reading inside the metric's declared band", async () => {
vi.mocked(fetchMeasurements).mockResolvedValue([
{
type: "PULSE",
value: 112,
measuredAt: new Date("2026-05-16T08:00:00Z"),
},
] as never);
const result = await eventStorage.run(event, () =>
syncUserMeasurements("user-1"),
);
expect(result).toEqual({ imported: 1, failed: false });
expect(prisma.measurement.create).toHaveBeenCalledTimes(1);
expect(tally()).toBeUndefined();
});
it("refuses an out-of-band reading and annotates the drop", async () => {
vi.mocked(fetchMeasurements).mockResolvedValue([
{
type: "PULSE",
value: 112,
measuredAt: new Date("2026-05-16T08:00:00Z"),
},
{
type: "PULSE",
value: 55643821.505,
measuredAt: new Date("2026-05-16T08:05:00Z"),
},
] as never);
const result = await eventStorage.run(event, () =>
syncUserMeasurements("user-1"),
);
// Only the real reading landed, and the run is still a clean success: one
// impossible sample must not strand the rest behind a held watermark.
expect(result).toEqual({ imported: 1, failed: false });
expect(prisma.measurement.create).toHaveBeenCalledTimes(1);
expect(tally()).toEqual({
total: 1,
buckets: [
{ source: "withings", type: "PULSE", direction: "above_max", count: 1 },
],
});
expect(event.getLevel()).toBe("warn");
});
it("does not overwrite a stored reading with an out-of-band one", async () => {
// The slot already holds a real pulse; the provider re-sends the same
// instant carrying a decode slip. Ungated, the update branch rewrote the
// good value with the bad one.
vi.mocked(prisma.measurement.findFirst).mockResolvedValue({
id: "existing-row",
} as never);
vi.mocked(fetchMeasurements).mockResolvedValue([
{
type: "PULSE",
value: 55643821.505,
measuredAt: new Date("2026-05-16T08:00:00Z"),
},
] as never);
const result = await eventStorage.run(event, () =>
syncUserMeasurements("user-1"),
);
expect(result).toEqual({ imported: 0, failed: false });
expect(prisma.measurement.update).not.toHaveBeenCalled();
expect(tally()?.total).toBe(1);
});
});
describe("syncUserActivity — plausibility gate", () => {
it("plans a daily total inside the metric's declared band", async () => {
installActivityFetch([{ date: "2026-05-16", steps: 8_400 }]);
const imported = await eventStorage.run(event, () =>
syncUserActivity("user-1"),
);
expect(imported).toBe(1);
expect(prisma.measurement.createMany).toHaveBeenCalledTimes(1);
expect(tally()).toBeUndefined();
});
it("refuses an out-of-band daily total and annotates the drop", async () => {
// ACTIVITY_STEPS is declared 0–200 000.
installActivityFetch([
{ date: "2026-05-16", steps: 8_400 },
{ date: "2026-05-17", steps: 9_900_000 },
]);
const imported = await eventStorage.run(event, () =>
syncUserActivity("user-1"),
);
expect(imported).toBe(1);
const created = vi.mocked(prisma.measurement.createMany).mock.calls[0]?.[0]
?.data as Array<{ value: number }>;
expect(created).toHaveLength(1);
expect(created[0]!.value).toBe(8_400);
expect(tally()).toEqual({
total: 1,
buckets: [
{
source: "withings",
type: "ACTIVITY_STEPS",
direction: "above_max",
count: 1,
},
],
});
expect(event.getLevel()).toBe("warn");
});
});
+12 -2
View File
@@ -39,6 +39,7 @@ import { prisma } from "@/lib/db";
import { getEvent } from "@/lib/logging/context";
import { safeFetch } from "@/lib/safe-fetch";
import { getUnitForType } from "@/lib/validations/measurement";
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
import {
collapseToTypeDayKeys,
recomputeBucketsForMeasurement,
@@ -342,7 +343,7 @@ export async function syncUserActivity(
value: number;
externalId: string;
}
const planned: PlannedRow[] = [];
const collected: PlannedRow[] = [];
for (const entry of entries) {
if (!entry.date) continue;
const measuredAt = activityMeasuredAt(entry.date);
@@ -352,7 +353,7 @@ export async function syncUserActivity(
// valid (a day of rest still records 0 steps). Only skip
// undefined / null / NaN.
if (raw == null || !Number.isFinite(raw)) continue;
planned.push({
collected.push({
type,
measuredAt,
value: raw,
@@ -361,6 +362,15 @@ export async function syncUserActivity(
}
}
// A daily-activity total the app's own band calls impossible (a step count
// past 200 000, a negative distance) is a provider artefact, not a day. Drop
// it before the plan is read against the table, so it can neither insert a
// fresh impossible slot nor overwrite a good stored one.
const planned = dropImplausibleMeasurements("withings", collected, (p) => ({
type: p.type,
value: p.value,
}));
if (planned.length > 0) {
const slotKey = (type: MeasurementType, measuredAt: Date) =>
`${type}|${measuredAt.getTime()}`;
+6
View File
@@ -474,6 +474,9 @@ export async function syncUserSleep(
if (verdict.status === "failed") {
throw new MeasurementReconciliationError(verdict);
}
// Refused by the plausibility gate: nothing was written and nothing
// was touched. The drop is already on the wide event.
if (verdict.status === "rejected_range") continue;
for (const dirty of verdict.dirtyIdentities ?? []) {
touched.push(dirty);
}
@@ -542,6 +545,9 @@ export async function syncUserSleep(
if (verdict.status === "failed") {
throw new MeasurementReconciliationError(verdict);
}
// Refused by the plausibility gate: nothing was written and
// nothing was touched. The drop is already on the wide event.
if (verdict.status === "rejected_range") continue;
for (const dirty of verdict.dirtyIdentities ?? []) {
touched.push(dirty);
}
+13 -2
View File
@@ -15,6 +15,7 @@ import { getUserWithingsCredentials } from "./credentials";
import { annotate, getEvent } from "@/lib/logging/context";
import { isP2002 } from "@/lib/prisma-errors";
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
import {
isReauthRequired,
recordSyncFailure,
@@ -229,14 +230,24 @@ export async function syncUserMeasurements(
)
: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); // 30 days back
let measures: Awaited<ReturnType<typeof fetchMeasurements>>;
let fetched: Awaited<ReturnType<typeof fetchMeasurements>>;
try {
measures = await fetchMeasurements(tokenInfo.accessToken, startDate);
fetched = await fetchMeasurements(tokenInfo.accessToken, startDate);
} catch (err) {
await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_MEASURES);
throw err;
}
// The app's own plausibility band applies to a provider reading exactly as
// it applies to a typed one. A remote that reports a pulse in the millions
// is reporting a decode slip, not a heartbeat, and the row would poison
// every mean and personal band derived from it. Refuse it here, before the
// update branch below can overwrite a good stored value with it.
const measures = dropImplausibleMeasurements("withings", fetched, (m) => ({
type: m.type,
value: m.value,
}));
let imported = 0;
// v1.4.39.1 — track every (type, measuredAt) we touched so the
// persistent rollup tier can be re-folded at the end of the sync.