diff --git a/src/__tests__/apple-health-unit-audit-guard.test.ts b/src/__tests__/apple-health-unit-audit-guard.test.ts index 070472f32..af341c9dc 100644 --- a/src/__tests__/apple-health-unit-audit-guard.test.ts +++ b/src/__tests__/apple-health-unit-audit-guard.test.ts @@ -48,7 +48,7 @@ import { convertHkValue, isConvertibleHkUnit, } from "@/lib/measurements/hk-units"; -import { readSource } from "@/__tests__/helpers/source-files"; +import { readSource, readSourceFolder } from "@/__tests__/helpers/source-files"; const ENTRIES = Object.values(APPLE_HEALTH_TYPE_MAP); @@ -194,10 +194,17 @@ describe("Apple Health mapping — the record's own unit (issue #944)", () => { "utf8", ); expect(archive).toContain(optIn); - const batch = readSource("src/app/api/measurements/batch/route.ts", [ - "export const POST", - "hkIdentifier", - ]); + // The route parses; its ingest maps the entries in the service folder. + const batch = [ + readSource("src/app/api/measurements/batch/route.ts", [ + "export const POST", + "hkIdentifier", + ]), + ...readSourceFolder("src/lib/measurements/batch-ingest", { + floor: 5, + }).map(([, text]) => text), + ].join("\n"); + expect(batch).toContain("mapAppleHealthEntry("); expect( batch, "the batch route opted into record-unit conversion", diff --git a/src/__tests__/dense-measurement-read-guard.test.ts b/src/__tests__/dense-measurement-read-guard.test.ts index 07c9d441b..086a54cb3 100644 --- a/src/__tests__/dense-measurement-read-guard.test.ts +++ b/src/__tests__/dense-measurement-read-guard.test.ts @@ -64,6 +64,9 @@ const SCOPED_DIRS = [ "lib/export", "lib/targets/", "lib/doctor-report", + // Request-handler logic moved out of its route into a service keeps the + // route's scope. + "lib/measurements/batch-ingest/", ]; /** Types a device writes at sampling rate rather than a few times a day. */ @@ -103,7 +106,7 @@ const ALLOWED: Record = { "Health-score inputs: steps (drained to one row per day nightly), sleep, waist, weight, blood pressure, fasting glucose only.", 'app/api/analytics/route.ts::"BLOOD_GLUCOSE"': "The 30-day per-context glucose summary, the same window and bound as the dashboard and coach glucose panels: at a sensor's fixed rate, below 45 000 rows.", - "app/api/measurements/batch/route.ts::p.row.type as MeasurementType": + "lib/measurements/batch-ingest/write-measurements.ts::p.row.type as MeasurementType": "Cross-source merge probe: one ±tolerance window around each posted reading's instant, so the rows found are bounded by the batch size.", "app/api/measurements/route.ts::(shorthand)": "Cross-source merge probe for one posted sleep segment: SLEEP_DURATION in a ±tolerance window around its end.", diff --git a/src/__tests__/external-id-stability-guard.test.ts b/src/__tests__/external-id-stability-guard.test.ts index 3d3d27425..10e1a9f40 100644 --- a/src/__tests__/external-id-stability-guard.test.ts +++ b/src/__tests__/external-id-stability-guard.test.ts @@ -394,17 +394,28 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => { // Each of these drains a client sync queue. A schema-level refusal here // would stop a whole batch of good rows because of one bad one, which is // worse than the duplicate it prevents. - const BATCH_ROUTES = [ - "app/api/measurements/batch/route.ts", - "app/api/mood-entries/bulk/route.ts", - "app/api/medications/intake/bulk/route.ts", - "app/api/workouts/batch/route.ts", - "app/api/cycle/day-logs/bulk/route.ts", - "app/api/import/route.ts", - ]; + // + // A surface whose ingest moved into a service names the service files + // beside the route: the route parses, the service refuses per entry. + const BATCH_SURFACES: Record = { + "app/api/measurements/batch/route.ts": [ + "lib/measurements/batch-ingest/prepare.ts", + "lib/measurements/batch-ingest/ingest.ts", + ], + "app/api/mood-entries/bulk/route.ts": [], + "app/api/medications/intake/bulk/route.ts": [], + "app/api/workouts/batch/route.ts": [], + "app/api/cycle/day-logs/bulk/route.ts": [], + "app/api/import/route.ts": [], + }; + const BATCH_ROUTES = Object.keys(BATCH_SURFACES); + + /** The route and the service files that carry its per-entry ingest. */ + const surface = (rel: string): string => + [rel, ...BATCH_SURFACES[rel]].map((file) => body(file)).join("\n"); it.each(BATCH_ROUTES)("%s classifies each entry itself", (rel) => { - const source = body(rel); + const source = surface(rel); expect(source).toContain("classifyExternalId"); expect(source).toContain("unstableExternalIdMeta"); }); @@ -414,7 +425,7 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => { (rel) => { // The JSON importer has no per-entry status envelope — it reports a // skipped COUNT — so it is the one batch surface without the code. - expect(body(rel)).toContain("UNSTABLE_EXTERNAL_ID_REASON"); + expect(surface(rel)).toContain("UNSTABLE_EXTERNAL_ID_REASON"); }, ); }); diff --git a/src/__tests__/measurement-write-plausibility-guard.test.ts b/src/__tests__/measurement-write-plausibility-guard.test.ts index d7a6d35a1..a26ebc4ca 100644 --- a/src/__tests__/measurement-write-plausibility-guard.test.ts +++ b/src/__tests__/measurement-write-plausibility-guard.test.ts @@ -73,10 +73,13 @@ const WRITERS: Record = { validatedIn: "lib/validations/measurement.ts", }, "app/api/import/route.ts": { kind: "validator" }, - // v1.44 — the set-based insert writes rows that already passed the - // route's `validateMeasurementRange` per entry; everything else goes - // through the reconciler, which gates itself. - "app/api/measurements/batch/route.ts": { kind: "validator" }, + // The batch ingest's set-based insert writes rows that already passed + // `validateMeasurementRange` per entry in its first pass; everything else + // goes through the reconciler, which gates itself. + "lib/measurements/batch-ingest/write-measurements.ts": { + kind: "upstream", + validatedIn: "lib/measurements/batch-ingest/prepare.ts", + }, "lib/mcp/writes.ts": { kind: "validator" }, "lib/measurements/create-from-telegram.ts": { kind: "validator" }, "lib/measurements/import-apple-health-export.ts": { kind: "validator" }, diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 1faa22114..24e03e639 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -32,6 +32,13 @@ const WRITE_RE = /** Runs `afterMeasurementMutation` on the rows it wrote. */ const TAIL = "runs afterMeasurementMutation"; +/** + * Writes here; the module named runs `afterMeasurementMutation` on the same + * rows (a service split into stages, whose write and tail are siblings). + */ +const TAIL_VIA_PREFIX = "runs afterMeasurementMutation via "; +const TAIL_VIA = (file: string) => `${TAIL_VIA_PREFIX}${file}`; + /** * Every file with a measurement write, and how it keeps the two readers * current. A file listed here must still contain a write (a stale entry @@ -44,7 +51,9 @@ const WRITE_SITES: Record = { "app/api/measurements/bulk-delete/route.ts": TAIL, "app/api/measurements/by-external-ids/route.ts": TAIL, "app/api/measurements/restore/route.ts": TAIL, - "app/api/measurements/batch/route.ts": TAIL, + "lib/measurements/batch-ingest/write-measurements.ts": TAIL_VIA( + "lib/measurements/batch-ingest/tail.ts", + ), "app/api/measurements/route.ts": TAIL, "app/api/mental-health/assessments/route.ts": TAIL, "lib/insights/score-row.ts": TAIL, @@ -157,8 +166,14 @@ describe("measurement write tail", () => { it("the files that claim the shared tail call it", () => { const missing = Object.entries(WRITE_SITES) - .filter(([, reason]) => reason === TAIL) - .map(([file]) => file) + .filter( + ([, reason]) => reason === TAIL || reason.startsWith(TAIL_VIA_PREFIX), + ) + .map(([file, reason]) => + reason.startsWith(TAIL_VIA_PREFIX) + ? reason.slice(TAIL_VIA_PREFIX.length) + : file, + ) .filter( (file) => !/afterMeasurementMutation\s*\(/.test( diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index d634884bc..2f3b785e0 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -36,15 +36,11 @@ * contract. */ import { NextRequest } from "next/server"; -import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { z } from "zod/v4"; -import { prisma } from "@/lib/db"; import { apiHandler, isScopedCredential, requireAuth } from "@/lib/api-handler"; -import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source"; import { MEASUREMENTS_WRITE_SCOPE } from "@/lib/measurements/scopes"; import { annotate } from "@/lib/logging/context"; -import { auditLog } from "@/lib/auth/audit"; import { apiError, apiSuccess, @@ -54,87 +50,13 @@ import { sanitiseZodIssues, } from "@/lib/api-response"; import { withIdempotency } from "@/lib/idempotency"; -import { - mapAppleHealthEntry, - isAggregatedBucketExternalId, - targetsAggregatedBucket, -} from "@/lib/measurements/apple-health-mapping"; -import { - MEASURED_AT_TOLERANCE_MS, - isMergeableSource, - isSameReadingAcrossSource, - oppositeMergeSource, - type MergeCandidate, -} from "@/lib/measurements/cross-source-merge"; -import { - externalKeyOf, - onceInBatch, - readStoredSampleIdentities, - readStoredStatsRows, - sampleIsStoredDuplicate, - statsWriteUnchanged, -} from "@/lib/measurements/batch-stored-rows"; -import { - lockMeasurementUser, - reconcileExternalMeasurement, - type MeasurementReconciliationVerdict, -} from "@/lib/measurements/reconcile-external-measurement"; -import { classifyRangeRejection } from "@/lib/measurements/plausibility-gate"; -import { - FOLDED_WINDOW_REASON, - findFoldedWindowDuplicates, -} from "@/lib/measurements/folded-window"; -import { - BACKFILL_SYNC_TRIGGER, - HEALTHKIT_REQUEST_SYNC_TRIGGERS, - recordHealthKitTypeArrivals, - tallyHealthKitArrivals, -} from "@/lib/integrations/healthkit-type-sync"; -import { - classifyHourBucket, - HOUR_BUCKET_SKIP_REASONS, - combineHourWrites, - parseHourBucketId, - writeHourBuckets, - type HourBucketEntry, - type HourBucketWrite, -} from "@/lib/measurements/intraday-hour-buckets"; -import { resolveStoredTimezone } from "@/lib/tz/resolver"; -import { isPlausibleEntryInstant } from "@/lib/validations/entry-instant"; -import { - classifyExternalId, - unstableExternalIdMeta, - UNSTABLE_EXTERNAL_ID_REASON, - type UnstableExternalIdShape, -} from "@/lib/validations/external-id"; -import { - VALUE_RANGES, - getUnitForType, - validateMeasurementRange, -} from "@/lib/validations/measurement"; +import type { BatchIngestEntry } from "@/lib/measurements/batch-ingest/entries"; +import { ingestMeasurementBatch } from "@/lib/measurements/batch-ingest/ingest"; +import { HEALTHKIT_REQUEST_SYNC_TRIGGERS } from "@/lib/integrations/healthkit-type-sync"; import { deviceTypeEnum } from "@/lib/validations/source-priority"; import { checkRateLimit } from "@/lib/rate-limit"; -import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; -import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; -import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; -import { emitDataArrival } from "@/lib/arrivals/emit-shared"; -import { groupRowsByArrivalKind } from "@/lib/arrivals/measurement-kind"; -import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; -import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; -import { - Prisma, - type MeasurementSource, - type MeasurementType, -} from "@/generated/prisma/client"; import { isoDateTime } from "@/lib/validations/datetime"; -// v1.4.25 — historical-backfill threshold for PR push -// suppression. A batch larger than this fires the detection job with -// `silent: true` so a multi-year Apple Health backfill writes records -// without spamming the user with hundreds of pushes. Tuned generously -// — a healthy daily-sync batch is typically well under 50 entries. -const PR_DETECTION_SILENT_THRESHOLD = 50; - const MAX_BATCH_ENTRIES = 500; // Body-size ceiling backstopping the per-entry cap. A worst-case entry @@ -146,12 +68,6 @@ const MAX_BATCH_ENTRIES = 500; // The cap is a DoS ceiling, not a tight bound. const MAX_BODY_BYTES = 4 * 1024 * 1024; -/** - * Rows per set-based insert statement. 500 entries fit two statements; each - * row binds about fifteen parameters, far under PostgreSQL's 65 535. - */ -const SET_INSERT_CHUNK = 250; - // v1.4.25 (security H-2): cap batch ingest at 60 // batches per user per minute. Healthy iOS sync drains its // HealthKit observer queue in well under one batch per minute (the @@ -197,7 +113,7 @@ const batchEntrySchema = z.object({ valueMax: z.number().finite().optional(), // Stability is enforced per entry in the loop below, not at the field: // a schema-level refusal would fail the whole batch on one bad row. - // @external-id-checked-per-entry: src/app/api/measurements/batch/route.ts + // @external-id-checked-per-entry: src/lib/measurements/batch-ingest/prepare.ts externalId: z.string().min(1).max(120), externalSourceVersion: z.string().min(1).max(120).optional(), // v1.8.6 — optional per-entry source tag. Defaults to @@ -238,71 +154,6 @@ const batchPayloadSchema = z.object({ syncTrigger: syncTriggerEnum.optional(), }); -type BatchEntry = z.infer; - -/** - * Per-entry outcome the iOS client uses to advance its sync cursor. - * `inserted` and `duplicate` both indicate the row landed (or was - * already present), so the client can checkpoint past them. `updated` - * is the per-day-aggregate overwrite path: when the iOS client re-posts - * a `stats:*` cumulative row (Steps, Active Energy, Sleep Duration, - * Walking/Running Distance, Flights Climbed — see issue #213) for a day - * we already have, the server overwrites the row's `value` rather than - * dropping the new payload as a duplicate. Sample-class rows (every - * other HK metric) keep the strict `duplicate` semantics because each - * sample is a canonical, immutable reading. `skipped` is a validation no-op; - * `failed` is a hard database verdict and must not be checkpointed as landed. - */ -type EntryStatus = "inserted" | "updated" | "duplicate" | "skipped" | "failed"; - -// v1.5.0 issue #213 — `stats:*` externalIds carry per-day cumulative -// totals that the iOS HealthKit observer re-posts as the day progresses. -// Treating those as immutable duplicates freezes the day's tile at the -// first-sync value. Sample-class externalIds (every other prefix) keep -// the strict insert-only contract. -const STATS_EXTERNAL_ID_PREFIX = "stats:"; -function isStatsExternalId(externalId: string | null | undefined): boolean { - return ( - typeof externalId === "string" && - externalId.startsWith(STATS_EXTERNAL_ID_PREFIX) - ); -} -interface EntryResult { - index: number; - status: EntryStatus; - reason?: string; - /** - * Only on a `value_out_of_range` skip: the value after conversion into the - * stored unit, which is what the band was checked against. - */ - convertedValue?: number; - /** Only on a `value_out_of_range` skip: the band applied, in the stored unit. */ - range?: { min: number; max: number; unit: string }; -} - -/** - * The `value_out_of_range` skip, with what a client needs to say which limit - * applied. The value is the entry's own reading after conversion, handed back - * to the account that sent it; a fraction sent where a percent was expected - * shows up here as a converted value a hundred times too small or too large. - */ -function outOfRangeResult( - index: number, - type: string, - convertedValue: number, -): EntryResult { - const band = VALUE_RANGES[type]; - return { - index, - status: "skipped", - reason: "value_out_of_range", - ...(Number.isFinite(convertedValue) ? { convertedValue } : {}), - ...(band - ? { range: { min: band.min, max: band.max, unit: getUnitForType(type) } } - : {}), - }; -} - export const POST = apiHandler(withIdempotency<[NextRequest]>(postBatch)); async function postBatch(request: NextRequest): Promise { @@ -406,1156 +257,17 @@ async function postBatch(request: NextRequest): Promise { ); } - const { entries, syncTrigger } = parsed.data; - - // First pass — map each inbound entry; record skipped entries. - type Prepared = { - index: number; - entry: BatchEntry; - row: Prisma.MeasurementCreateManyInput; - }; - const prepared: Prepared[] = []; - const results: EntryResult[] = new Array(entries.length); - const unstableShapes: UnstableExternalIdShape[] = []; - // v1.43 — hourly sums for the intraday profile, and the zone they are cut in - // (the account's, the one `/api/auth/me` publishes as `timezone`). - const hourWrites: HourBucketEntry[] = []; - let accountTz: string | undefined; - const batchNow = new Date(); - - for (let index = 0; index < entries.length; index++) { - const entry = entries[index]; - - // The `(user_id, type, source, external_id)` dedup key above is only - // idempotent while the id is STABLE across client launches. An id - // that rotates per process (an object description carrying a memory - // address) never matches its own earlier row, so every sync sweep - // mints a fresh measurement. Refused per entry, like the range and - // timestamp guards below — one malformed row must not stop the other - // 499 from landing. - const unstable = classifyExternalId(entry.externalId); - if (unstable) { - unstableShapes.push(unstable); - results[index] = { - index, - status: "skipped", - reason: UNSTABLE_EXTERNAL_ID_REASON, - }; - continue; - } - - // v1.43 (ios #77) — an hourly sum of a summed type - // (`stats::T`) is never a measurement: it lands in the - // intraday profile behind the same-time baseline, and the day's total - // stays the day row. Validated here, written after the measurements. - const hourId = parseHourBucketId(entry.externalId); - if (hourId) { - const skip = (reason: string) => { - results[index] = { index, status: "skipped", reason }; - }; - // The profile has no source dimension and is the phone's: a bridge or a - // hand-entered row cannot post hours into it. - if (scoped || (entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") { - skip(HOUR_BUCKET_SKIP_REASONS.unsupportedType); - continue; - } - accountTz ??= await resolveStoredTimezone(user.timezone); - const startDate = new Date(entry.startDate); - const verdict = classifyHourBucket( - hourId, - startDate, - accountTz, - batchNow, - ); - if (!verdict.ok) { - skip(verdict.reason); - continue; - } - if (!isPlausibleEntryInstant(startDate)) { - skip("implausible_timestamp"); - continue; - } - const hourMapped = mapAppleHealthEntry({ - hkIdentifier: entry.hkIdentifier, - value: entry.value, - unit: entry.unit, - startDate: entry.startDate, - endDate: entry.endDate, - }); - if (!hourMapped || hourMapped.type !== verdict.type) { - // The entry's identifier must be the one its id names. - skip( - hourMapped - ? HOUR_BUCKET_SKIP_REASONS.unsupportedType - : "unmappable_identifier", - ); - continue; - } - if ( - hourMapped.value < 0 || - validateMeasurementRange(hourMapped.type, hourMapped.value) !== null - ) { - results[index] = outOfRangeResult( - index, - hourMapped.type, - hourMapped.value, - ); - continue; - } - hourWrites.push({ - index, - type: verdict.type, - hkIdentifier: hourId.hkIdentifier, - dateKey: hourId.dateKey, - hour: hourId.hour, - value: hourMapped.value, - startMs: startDate.getTime(), - }); - continue; - } - - const mapped = mapAppleHealthEntry({ - hkIdentifier: entry.hkIdentifier, - value: entry.value, - unit: entry.unit, - startDate: entry.startDate, - endDate: entry.endDate, - sleepStage: entry.sleepStage, - categoryValue: entry.categoryValue, - valueMin: entry.valueMin, - valueMax: entry.valueMax, - }); - - if (!mapped) { - results[index] = { - index, - status: "skipped", - reason: "unmappable_identifier", - }; - continue; - } - - // Plausibility-range guard. We tolerate "out of plausible range" - // by skipping rather than failing the batch — a single bogus - // sensor reading shouldn't poison an otherwise clean ingest. - const rangeError = validateMeasurementRange(mapped.type, mapped.value); - if (rangeError !== null) { - results[index] = outOfRangeResult(index, mapped.type, mapped.value); - continue; - } - - // Time-plausibility guard, the bound the single-entry twin already applies - // through `validateEntryInstant` on `createMeasurementSchema.measuredAt`. - // Without it a future-dated row permanently won "latest reading" on every - // dashboard and poisoned the rollup buckets and the canonical picker. The - // past floor stays at 1900 (no `maxAgeMs`) because a historical Apple - // Health backfill is a legitimate, expected shape on this route; only the - // future side is a bug. Skipped per entry, like the range guard above. - if (!isPlausibleEntryInstant(mapped.takenAt)) { - results[index] = { - index, - status: "skipped", - reason: "implausible_timestamp", - }; - continue; - } - - // v1.30.7 (iOS #34) — go-forward aggregated heart-rate wire - // contract. A `stats:HKQuantityTypeIdentifierHeartRate:<10-min-ISO-Z>` - // row is the 10-minute-average PULSE bucket; it rides the generic - // `stats:` overwrite path below so a within-bucket re-post (the running - // mean shifts) replaces the value instead of duplicating. Reject a row - // that targets a bucket prefix but carries a malformed suffix — a - // garbage suffix would mint an un-overwriteable row the next re-post - // can't collapse onto. A well-formed bucket (and every non-bucket - // `stats:` / per-sample `uuid` externalId) passes through unchanged. - if ( - targetsAggregatedBucket(entry.externalId) && - !isAggregatedBucketExternalId(entry.externalId) - ) { - results[index] = { - index, - status: "skipped", - reason: "malformed_hr_bucket_id", - }; - continue; - } - - // v1.30.7 (iOS #34) — the per-bucket MIN / MAX are persisted ONLY on a - // well-formed 10-min aggregated bucket row. A per-sample reading, a - // per-day cumulative `stats:` total, or a manual entry never carries a - // spread, so we pin them to null there even if a client mis-sends the - // fields — `value` stays the single source of truth for those rows. - // - // v1.30.8 — validate the spread the same way `value` is guarded (above): - // it now drives the DAY min/max band + the intraday envelope, so an - // out-of-range or mis-ordered spread (a spurious `discreteMin` of 0, a - // sensor glitch) would poison the aggregate exactly like a bogus `value` - // would. Persist the spread ONLY when both extremes are present, in range, - // and correctly ordered around `value`; otherwise drop to null and keep the - // trustworthy average. Partial or invalid spread → avg-only, still valid. - const isHrBucket = isAggregatedBucketExternalId(entry.externalId); - const spreadValid = - isHrBucket && - typeof mapped.valueMin === "number" && - typeof mapped.valueMax === "number" && - validateMeasurementRange(mapped.type, mapped.valueMin) === null && - validateMeasurementRange(mapped.type, mapped.valueMax) === null && - mapped.valueMin <= mapped.value && - mapped.value <= mapped.valueMax; - const valueMin = spreadValid ? mapped.valueMin : null; - const valueMax = spreadValid ? mapped.valueMax : null; - - prepared.push({ - index, - entry, - row: { - userId: user.id, - type: mapped.type, - value: mapped.value, - valueMin, - valueMax, - unit: mapped.unit, - // v1.8.6 — honour the per-entry source tag, defaulting to - // `APPLE_HEALTH` when absent so legacy callers are unchanged. - // - // v1.38.x — a scoped credential gets `EXTERNAL`, and can hold no other - // value: the guard above refused any body that named one, so nothing - // asserted is being discarded here. The value has to move with the - // credential rather than with the route, because the route serves both - // the phone and a third-party bridge. It also moves the dedup key to - // `(userId, type, EXTERNAL, externalId)`, which is the isolated - // namespace this source exists to give. - source: scoped ? EXTERNAL_SOURCE : (entry.source ?? "APPLE_HEALTH"), - measuredAt: mapped.takenAt, - externalId: entry.externalId, - externalSourceVersion: entry.externalSourceVersion ?? null, - sleepStage: mapped.sleepStage ?? null, - // v1.10.0 — categorical events (WX-B). The device's own verdict / - // severity for an EVENT row; null for every continuous reading. - rhythmClassification: mapped.rhythmClassification ?? null, - // v1.4.25 — pass the iOS-supplied device-type through to - // the row. Stays null for pre-W8c iOS builds; the canonical - // picker treats null as `unknown` so legacy ingest keeps - // working without a server-side default. - deviceType: entry.deviceType ?? null, - }, - }); - } - - if (unstableShapes.length > 0) { - annotate({ - meta: unstableExternalIdMeta("measurement.batch", unstableShapes), - }); - } - - let insertedCount = 0; - let updatedCount = 0; - /** - * `stats:` re-posts reported `updated` with nothing to write (see - * `isUnchangedStatsRow`). Counted in the response's `updated` like before, - * and in nothing that starts work. - */ - let unchangedStatsCount = 0; - const unchangedStatsRows = new Set(); // #1259: not an arrival - let duplicateCount = 0; - let failedCount = 0; - // v1.44 — subset of `insertedCount` written by the set-based insert. - let setInsertedCount = 0; - // v1.11.4 (iOS #2) — subset of `duplicateCount` collapsed by the - // MANUAL↔APPLE_HEALTH same-reading merge (as opposed to a plain - // composite-key duplicate). Surfaced as a dedicated wide-event count so - // an operator can see how often the standalone-pair mirror collapses. - let crossSourceMergedCount = 0; - // v1.42 — subset of `duplicateCount` refused by the `folded_window` guard. - let foldedWindowCount = 0; - // Only rows the reconciler actually inserted drive arrival side effects. - const insertedPrepared: Prepared[] = []; - const writtenIdentities: Array<{ - type: MeasurementType; - measuredAt: Date; - }> = []; - - if (prepared.length > 0) { - // v1.11.4 (iOS #2) — same-reading merge for the MANUAL ↔ APPLE_HEALTH - // mirror duplicate. When the iOS app logs a manual reading offline it - // mirrors that reading into HealthKit; on pairing, adopt-on-pair - // uploads it as MANUAL and HealthKit background sync independently - // re-ingests the mirrored sample as APPLE_HEALTH. The two carry - // different externalIds + sources, so the composite unique index lets - // both land — duplicating the whole hand-logged history on first pair. - // We collapse them at ingest: an incoming MANUAL / APPLE_HEALTH row is - // dropped as a `duplicate` when a same-reading row of the OPPOSITE - // source already exists (same type + value + measuredAt within a tight - // ±2 s window). `stats:*` cumulative rows are excluded — those are - // per-day aggregates the observer overwrites in place, never - // hand-entered point readings. See `cross-source-merge.ts` for the - // full rule + why first-physical-reading-wins (the value is identical - // between the two rows, so only the source label differs). - // - // Pull the opposite-source candidate rows in one widened query: for - // every mergeable, non-`stats:*` incoming row, look for an opposite- - // source row of the same type whose measuredAt falls inside the - // tolerance window. Compared in JS because the value match needs a - // float epsilon the SQL `IN` can't express. - const mergeProbes = prepared.filter( - (p) => - isMergeableSource(p.row.source as string) && - !isStatsExternalId(p.row.externalId as string), - ); - const crossSourceCandidates: MergeCandidate[] = []; - if (mergeProbes.length > 0) { - const candidateRows = await prisma.measurement.findMany({ - where: { - userId: user.id, - deletedAt: null, - OR: mergeProbes.map((p) => { - const measuredAt = p.row.measuredAt as Date; - return { - type: p.row.type as MeasurementType, - source: oppositeMergeSource( - p.row.source as "MANUAL" | "APPLE_HEALTH", - ), - measuredAt: { - gte: new Date(measuredAt.getTime() - MEASURED_AT_TOLERANCE_MS), - lte: new Date(measuredAt.getTime() + MEASURED_AT_TOLERANCE_MS), - }, - }; - }), - }, - select: { type: true, source: true, value: true, measuredAt: true }, - }); - crossSourceCandidates.push(...candidateRows); - } - - // Rows already chosen for insert earlier in THIS batch — so a MANUAL - // and an APPLE_HEALTH same-reading arriving in the SAME batch also - // collapse (the DB probe above only sees rows from prior batches). - const inBatchCandidates: MergeCandidate[] = []; - function hasSameReadingSibling(p: Prepared): boolean { - const incoming = { - type: p.row.type as MeasurementType, - source: p.row.source as string, - value: p.row.value as number, - measuredAt: p.row.measuredAt as Date, - }; - if (!isMergeableSource(incoming.source)) return false; - for (const candidate of crossSourceCandidates) { - if (isSameReadingAcrossSource(incoming, candidate)) return true; - } - for (const candidate of inBatchCandidates) { - if (isSameReadingAcrossSource(incoming, candidate)) return true; - } - return false; - } - - // `stats:` rows carry overwrite semantics and the last entry in one batch - // is authoritative. Earlier snapshots are safe duplicates. - const lastStatsIndexByKey = new Map(); - for (const p of prepared) { - if (!isStatsExternalId(p.row.externalId as string)) continue; - lastStatsIndexByKey.set( - `${p.row.type}::${p.row.source}::${p.row.externalId}`, - p.index, - ); - } - - // v1.42 — exact-duplicate prefilter. A re-sent sample used to cost one - // interactive transaction of about ten round-trips (savepoint, three - // advisory locks, two candidate reads, a row lock) only for the - // reconciler to answer `duplicate` without writing. A backfill after an - // anchor reset re-sends whole 500-entry batches of exactly that. - // - // The reconciler answers `duplicate` for a non-`stats:` row exactly when - // an existing row, live OR tombstoned, has the same - // `(type, source, externalId)` AND the same `measuredAt` and - // `sleepStage`: the natural-identity index is unique NULLS NOT DISTINCT, - // so that row is both its external and its natural match, and with no - // aggregation provenance on a sample nothing else can win. One read over - // the batch's sample ids, without a `deletedAt` filter, answers that for - // every row at once. - // - // The verdict is taken at the very point the reconciler would have been - // called, after `superseded_in_batch` and the cross-source merge, so - // statuses, reasons, counters and the in-batch merge candidates are the - // same as before. A row is only prefiltered when neither its external - // key nor its natural identity occurs a second time in the batch: an - // earlier entry sharing either could move or retire the matched row - // before this one is reached, and then the reconciler has to decide. - // `stats:` rows are never prefiltered (overwrite semantics), and neither - // is a value the reconciler's own plausibility gate would refuse. - const onceInThisBatch = onceInBatch(prepared.map((p) => p.row)); - const existingByExternalKey = await readStoredSampleIdentities( - prisma, - user.id, - [ - ...new Set( - prepared - .map((p) => p.row.externalId as string) - .filter((externalId) => !isStatsExternalId(externalId)), - ), - ], - ); - const isExactStoredDuplicate = (p: Prepared): boolean => - sampleIsStoredDuplicate( - existingByExternalKey.get( - externalKeyOf(p.row.type, p.row.source, p.row.externalId), - ), - p.row, - ) && - onceInThisBatch(p.row) && - classifyRangeRejection( - p.row.type as MeasurementType, - p.row.value as number, - ) === null; - - // An unchanged `stats:` re-post is reported `updated` without a write - // (see `batch-stored-rows.ts`), under the same once-per-batch key rule. - const storedStats = await readStoredStatsRows(prisma, user.id, [ - ...new Set( - prepared - .map((p) => p.row.externalId as string) - .filter((externalId) => isStatsExternalId(externalId)), - ), - ]); - const isUnchangedStatsRow = (p: Prepared): boolean => - onceInThisBatch(p.row) && - classifyRangeRejection( - p.row.type as MeasurementType, - p.row.value as number, - ) === null && - statsWriteUnchanged( - storedStats.get( - externalKeyOf(p.row.type, p.row.source, p.row.externalId), - ), - desiredFor(p), - ); - - // v1.42 — the `folded_window` guard (`folded-window.ts`). A raw Apple - // Health sample of a dense or mean type, old enough to have been folded, - // whose hour or day a live `stats:` row already covers, is a duplicate: - // the fold no longer leaves the raw rows behind as tombstones, so this is - // what keeps a re-upload after an anchor reset from standing beside the - // mean that already accounts for it. One read for the batch, and none at - // all when no row is old enough. - const foldedIndexes = await findFoldedWindowDuplicates( - prisma, - user.id, - prepared.map((p) => ({ - type: p.row.type as MeasurementType, - source: p.row.source as MeasurementSource, - externalId: p.row.externalId as string, - measuredAt: p.row.measuredAt as Date, - })), - ); - const foldedPrepared = new Set( - [...foldedIndexes].map((index) => prepared[index]), - ); - - /** The reconciler's input for one prepared row. */ - const desiredFor = (p: Prepared) => { - const statsRow = isStatsExternalId(p.row.externalId as string); - return { - userId: user.id, - type: p.row.type as MeasurementType, - value: p.row.value as number, - valueMin: p.row.valueMin as number | null, - valueMax: p.row.valueMax as number | null, - unit: p.row.unit as string, - source: p.row - .source as Prisma.MeasurementUncheckedCreateInput["source"], - measuredAt: p.row.measuredAt as Date, - externalId: p.row.externalId as string, - externalSourceVersion: p.row.externalSourceVersion as string | null, - glucoseContext: p.row - .glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"], - sleepStage: p.row - .sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"], - rhythmClassification: p.row - .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"], - deviceType: p.row.deviceType as string | null, - aggregationProvenance: statsRow - ? ("HEALTHKIT_STATISTICS" as const) - : null, - }; - }; - - const pushMergeCandidate = (p: Prepared) => { - if (isMergeableSource(p.row.source as string)) { - inBatchCandidates.push({ - type: p.row.type as MeasurementType, - source: p.row.source as string, - value: p.row.value as number, - measuredAt: p.row.measuredAt as Date, - }); - } - }; - - /** - * Record one row's verdict. `candidate` is false for a row whose merge - * candidate was already pushed when it was buffered (see below). - */ - const applyVerdict = ( - p: Prepared, - verdict: Pick & - Partial<{ - dirtyIdentities: Array<{ type: MeasurementType; measuredAt: Date }>; - error: { code?: string }; - }>, - candidate: boolean, - ) => { - switch (verdict.status) { - case "inserted": - results[p.index] = { index: p.index, status: "inserted" }; - insertedCount++; - insertedPrepared.push(p); - writtenIdentities.push({ - type: p.row.type as MeasurementType, - measuredAt: p.row.measuredAt as Date, - }); - break; - case "updated": - case "resurrected": - results[p.index] = { index: p.index, status: "updated" }; - updatedCount++; - writtenIdentities.push( - { - type: p.row.type as MeasurementType, - measuredAt: p.row.measuredAt as Date, - }, - ...(verdict.dirtyIdentities ?? []), - ); - break; - case "duplicate": - results[p.index] = { index: p.index, status: "duplicate" }; - duplicateCount++; - break; - case "failed": - results[p.index] = { - index: p.index, - status: "failed", - reason: verdict.error?.code ?? "write_failed", - }; - failedCount++; - return; - 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] = outOfRangeResult( - p.index, - p.row.type as string, - p.row.value as number, - ); - return; - } - if (candidate) pushMergeCandidate(p); - }; - - const reconcileRow = (p: Prepared) => - prisma.$transaction((tx) => - reconcileExternalMeasurement(tx, desiredFor(p), { - exactExternalMatch: isStatsExternalId(p.row.externalId as string) - ? "update" - : "duplicate", - }), - ); - - // v1.44 — set-based inserts. A fresh sample used to cost one interactive - // transaction of about ten round trips (savepoint, three advisory locks, - // two candidate reads, the insert), so a 500-entry batch of new samples - // was 500 of those in a row: the 2–15 s outliers. - // - // The reconciler inserts exactly when no row, live or tombstoned, holds - // the entry's external identity `(type, source, externalId)` or its - // natural identity `(type, measuredAt, source, sleepStage)`. Both are - // full unique indexes (the natural one NULLS NOT DISTINCT), so - // `INSERT … ON CONFLICT DO NOTHING` inserts exactly those rows and skips - // every other; a skipped row then goes through the reconciler as before - // and gets its own verdict (update, resurrect, duplicate). The insert - // runs under the same per-account advisory lock the reconciler takes, so - // the two serialise like two reconcilers would. - // - // Only a row whose external and natural key each occur once in the batch - // is buffered, and the buffer is flushed before any other row reaches the - // reconciler. Such a row shares no identity with another entry, so no - // other entry's reconcile can create, move or retire a row at its - // identity, and inserting it out of turn changes nothing. Rows the insert - // skips are reconciled in their original order. A buffered row's merge - // candidate is pushed when it is buffered, as the sequential loop pushed - // it after a successful write; the only case that differs is a write that - // fails outright, where the candidate stays. - const pendingInserts: Prepared[] = []; - const flushPendingInserts = async () => { - if (pendingInserts.length === 0) return; - const batch = pendingInserts.splice(0, pendingInserts.length); - const landed = new Set(); - try { - for (let i = 0; i < batch.length; i += SET_INSERT_CHUNK) { - const slice = batch.slice(i, i + SET_INSERT_CHUNK); - const rows = await prisma.$transaction(async (tx) => { - await lockMeasurementUser(tx, user.id); - return tx.measurement.createManyAndReturn({ - data: slice.map((p) => ({ - ...desiredFor(p), - sleepStage: (p.row.sleepStage ?? - null) as Prisma.MeasurementUncheckedCreateInput["sleepStage"], - deletedAt: null, - })), - skipDuplicates: true, - select: { type: true, source: true, externalId: true }, - }); - }); - const keys = new Set( - rows.map((r) => `${r.type}::${r.source}::${r.externalId}`), - ); - for (const p of slice) { - if (keys.has(`${p.row.type}::${p.row.source}::${p.row.externalId}`)) - landed.add(p); - } - } - } catch (err) { - // Whatever did not land goes through the reconciler below, which - // turns a real database fault into a per-entry `failed` verdict. - annotate({ - action: { name: "measurement.batch.set_insert_fallback" }, - meta: { - error_class: err instanceof Error ? err.name : "unknown", - landed: landed.size, - pending: batch.length, - }, - }); - } - setInsertedCount += landed.size; - for (const p of batch) { - if (landed.has(p)) applyVerdict(p, { status: "inserted" }, false); - else applyVerdict(p, await reconcileRow(p), false); - } - }; - - for (const p of prepared) { - const externalId = p.row.externalId as string; - const statsRow = isStatsExternalId(externalId); - const key = `${p.row.type}::${p.row.source}::${externalId}`; - if (statsRow && lastStatsIndexByKey.get(key) !== p.index) { - results[p.index] = { - index: p.index, - status: "duplicate", - reason: "superseded_in_batch", - }; - duplicateCount++; - continue; - } - - if (statsRow && isUnchangedStatsRow(p)) { - results[p.index] = { index: p.index, status: "updated" }; - unchangedStatsCount++; - unchangedStatsRows.add(p); - pushMergeCandidate(p); - continue; - } - - if (!statsRow && hasSameReadingSibling(p)) { - results[p.index] = { - index: p.index, - status: "duplicate", - reason: "cross_source_merge", - }; - duplicateCount++; - crossSourceMergedCount++; - continue; - } - - // Both verdicts below are the reconciler's `duplicate`, reached without - // its transaction; like every duplicate they fall through to the - // in-batch merge candidates. - const prefiltered = !statsRow && isExactStoredDuplicate(p); - const folded = !prefiltered && !statsRow && foldedPrepared.has(p); - if (prefiltered || folded) { - results[p.index] = folded - ? { - index: p.index, - status: "duplicate", - reason: FOLDED_WINDOW_REASON, - } - : { index: p.index, status: "duplicate" }; - duplicateCount++; - if (folded) foldedWindowCount++; - pushMergeCandidate(p); - continue; - } - - if (onceInThisBatch(p.row)) { - pendingInserts.push(p); - pushMergeCandidate(p); - continue; - } - - await flushPendingInserts(); - applyVerdict(p, await reconcileRow(p), true); - } - await flushPendingInserts(); - } - - // v1.43 — the hours, after the measurements. One write per slot: the last - // entry for one start wins (an earlier one in the same batch is a - // superseded duplicate), and the two starts of the hour a fall-back day - // repeats are summed (`combineHourWrites`). - let hourInsertedCount = 0; - let hourUpdatedCount = 0; - const hourLanded: HourBucketWrite[] = []; - const hourChanged = new Set(); // #1259: a new value - if (hourWrites.length > 0 && accountTz) { - const { writes, superseded } = combineHourWrites(hourWrites); - for (const index of superseded) { - results[index] = { - index, - status: "duplicate", - reason: "superseded_in_batch", - }; - duplicateCount++; - } - const toWrite = writes.map(({ write }) => write); - try { - const statuses = await writeHourBuckets( - prisma, - user.id, - accountTz, - toWrite, - ); - for (const { write, contributors } of writes) { - const status = statuses.get(write.index) ?? "inserted"; - // An unchanged hour still reads `updated` on the wire (#1259). - const wire = status === "inserted" ? "inserted" : "updated"; - for (const index of contributors) { - results[index] = { index, status: wire }; - if (status === "inserted") hourInsertedCount++; - else hourUpdatedCount++; - } - hourLanded.push(write); - if (status !== "unchanged") hourChanged.add(write); - } - } catch (err) { - for (const { contributors } of writes) { - for (const index of contributors) { - results[index] = { - index, - status: "failed", - reason: "write_failed", - }; - failedCount++; - } - } - annotate({ - action: { name: "measurement.batch.hour-bucket.failed" }, - meta: { error: err instanceof Error ? err.message : String(err) }, - }); - } - annotate({ - action: { name: "measurement.batch.hour-bucket" }, - meta: { - attempted: hourWrites.length, - inserted: hourInsertedCount, - updated: hourUpdatedCount, - processed: entries.length, - }, - }); - } - - const healthKitSyncSucceeded = - failedCount === 0 && - (hourLanded.length > 0 || - prepared.some((p) => { - if ((p.entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") return false; - const status = results[p.index]?.status; - return ( - status === "inserted" || - status === "updated" || - status === "duplicate" - ); - })); - - // This batch route is the native ingestion boundary, so it is also where the - // SyncMode checkpoint (`User.lastSyncedAt`, the same column - // `markSyncCheckpoint` writes) is established — `GET /api/sync/state` only - // reports it. The checkpoint covers MANUAL pushes too: the client synced - // whatever it was holding. A batch carrying a hard failure leaves the - // checkpoint where it was, so the undelivered window stays inside whatever - // the client asks for next. - // - // The HealthKit fields ride the same update and keep their narrower - // condition. Stamp only after every attempted write has reached a durable, - // non-failed verdict; MANUAL-only and skipped-only batches are not - // HealthKit syncs. - // - // The trigger the client declared is stamped alongside, and a background- or - // push-triggered batch additionally dates `healthKitLastBackgroundSyncAt`. - // Without that second timestamp there is no way to tell a phone that delivers - // on its own from one that only ever delivers while the app is open — both - // look like a moving `healthKitLastSyncedAt`. A batch that declares no - // trigger writes null over the previous value rather than leaving a stale one - // standing: the field describes THIS sync, and an older build's silence is - // not evidence about it. - // - // None of that is true of a scoped credential, which is why it is excluded - // from the whole block rather than only from the HealthKit half. The - // checkpoint's meaning is "the client synced what it was holding", and a - // third-party bridge is not the client and holds nothing on its behalf. - // `lastSyncedAt` is what `GET /api/sync/state` reports and what the phone - // uses to size its undelivered window, so letting a scale advance it can make - // the phone skip a window of its own samples — a silent data loss whose cause - // would be nowhere near the code that caused it. - // - // v1.43 (ios #77) — a `backfill` batch is a catch-up, not a sync. It records - // its types under `lastBackfillAt` only and moves none of the live stamps: - // thirty days of history must not read as thirty healthy daily syncs. - const backfill = syncTrigger === BACKFILL_SYNC_TRIGGER; - if (failedCount === 0 && !scoped) { - const syncedAt = new Date(); - const deliveredWithoutTheApp = - syncTrigger === "background" || syncTrigger === "push"; - // v1.42 (#1173) — per-type arrivals for the Apple Health card, Apple - // Health rows and hours only, under the checkpoint's conditions. Since - // #1259 only a new or changed one brings something (`brought`). - const arrivals = tallyHealthKitArrivals([ - ...prepared.flatMap((p) => { - const status = results[p.index]?.status; - if (p.row.source !== "APPLE_HEALTH" || !status) return []; - if (status === "skipped" || status === "failed") return []; - const unchanged = unchangedStatsRows.has(p); - const brought = status !== "duplicate" && !unchanged; - return [{ type: p.row.type as MeasurementType, brought }]; - }), - ...hourLanded.map((w) => ({ type: w.type, brought: hourChanged.has(w) })), - ]); - if (arrivals.size > 0) { - await recordHealthKitTypeArrivals(prisma, { - userId: user.id, - counts: arrivals, - trigger: syncTrigger ?? null, - at: syncedAt, - }); - } - if (!backfill) { - await prisma.user.update({ - where: { id: user.id }, - data: { - lastSyncedAt: syncedAt, - ...(healthKitSyncSucceeded - ? { - healthKitLastSyncedAt: syncedAt, - healthKitLastSyncTrigger: syncTrigger ?? null, - ...(deliveredWithoutTheApp - ? { healthKitLastBackgroundSyncAt: syncedAt } - : {}), - } - : {}), - }, - }); - } - } - - const freshlyInsertedRows = insertedPrepared.map((p) => ({ - type: p.row.type as MeasurementType, - measuredAt: - p.row.measuredAt instanceof Date - ? p.row.measuredAt - : new Date(p.row.measuredAt as string), - })); - - const skipped = results - .filter((r) => r.status === "skipped") - .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); - - // v1.32.1 (issue #585) — per-reason skip breakdown. The baseline - // annotation below only ever carried the total `skipped` count, so an - // operator investigating "why does this account's uploaded data look - // sparser than expected" had no way to see WHICH validation gate an - // entry tripped (malformed bucket id, out-of-range value, unmappable - // identifier, …) without pulling raw rows. Grouping by reason turns - // that into a one-glance wide-event field. - const skippedByReason: Record = {}; - for (const s of skipped) { - skippedByReason[s.reason] = (skippedByReason[s.reason] ?? 0) + 1; - } - - await auditLog("measurement.batch.ingest", { - userId: user.id, + const entries: BatchIngestEntry[] = parsed.data.entries; + const result = await ingestMeasurementBatch({ + user: { id: user.id, timezone: user.timezone }, + scoped, + entries, + syncTrigger: parsed.data.syncTrigger, ipAddress: getClientIp(request), - details: { - processed: entries.length, - inserted: insertedCount, - updated: updatedCount + unchangedStatsCount, - duplicates: duplicateCount, - skipped: skipped.length, - failed: failedCount, - }, }); - // v1.5.0 issue #213 — dedicated wide-event annotation so an operator - // can grep `measurement.batch.stats-overwrite` to see how often - // per-day cumulative rows are getting re-posted (a healthy ingest - // flow). Only fires when at least one row was overwritten so the - // baseline ingest trace is unchanged for batches with no `stats:*` - // duplicates. - if (updatedCount > 0) { - annotate({ - action: { name: "measurement.batch.stats-overwrite" }, - meta: { - updated: updatedCount, - processed: entries.length, - }, - }); - } - - // v1.11.4 (iOS #2) — dedicated annotation for the MANUAL↔APPLE_HEALTH - // same-reading collapse so an operator can grep - // `measurement.batch.cross-source-merge` to confirm the standalone-pair - // mirror duplicate is being absorbed (a healthy first-pair flow). Only - // fires when at least one row was merged so the baseline ingest trace - // is unchanged for the common case. - if (crossSourceMergedCount > 0) { - annotate({ - action: { name: "measurement.batch.cross-source-merge" }, - meta: { - merged: crossSourceMergedCount, - processed: entries.length, - }, - }); - } - - // v1.32.1 (issue #585) — dedicated outcome breakdown for entries - // targeting the go-forward aggregated 10-min heart-rate bucket prefix - // (`stats:HKQuantityTypeIdentifierHeartRate:<10-min-Z>`). The intraday - // pulse chart reads exactly these rows for a live-synced day; a report - // of "the chart looks sparser after the ZIP import cutoff" is - // unanswerable from the baseline ingest trace alone, since it never - // distinguished an HR-bucket write from any other entry in the batch. - // Grepping `measurement.batch.hr-bucket` now shows, per sync, how many - // bucket entries the client sent and what happened to each — confirming - // (or ruling out) server-side rejection/dedup as the density-gap cause. - // Only fires when the batch actually carried at least one HR-bucket - // entry so the baseline trace is unchanged for every other sync. - let hrBucketInserted = 0; - let hrBucketUpdated = 0; - let hrBucketDuplicate = 0; - let hrBucketSkipped = 0; - let hrBucketFailed = 0; - for (let index = 0; index < entries.length; index++) { - if (!targetsAggregatedBucket(entries[index].externalId)) continue; - switch (results[index]?.status) { - case "inserted": - hrBucketInserted++; - break; - case "updated": - hrBucketUpdated++; - break; - case "duplicate": - hrBucketDuplicate++; - break; - case "skipped": - hrBucketSkipped++; - break; - case "failed": - hrBucketFailed++; - break; - } - } - const hrBucketAttempted = - hrBucketInserted + - hrBucketUpdated + - hrBucketDuplicate + - hrBucketSkipped + - hrBucketFailed; - if (hrBucketAttempted > 0) { - annotate({ - action: { name: "measurement.batch.hr-bucket" }, - meta: { - attempted: hrBucketAttempted, - inserted: hrBucketInserted, - updated: hrBucketUpdated, - duplicate: hrBucketDuplicate, - skipped: hrBucketSkipped, - failed: hrBucketFailed, - processed: entries.length, - }, - }); - } - - // v1.4.25 — kick off PR detection for this user when at least one - // row was written, so a single off-day reading still gets evaluated; the - // warm-up gate inside the detector decides whether it's a record. Suppress - // push notifications for historical backfills above the silent threshold. - // Updated `stats:*` rows also count — a per-day-total overwrite can flip - // the day's value past a personal record. - // - // v1.42 — a batch of duplicates only no longer enqueues. It changed - // nothing a record or a reminder could read, and an anchor-reset backfill - // sends hundreds of them, each of which used to queue a detection pass, a - // reminder sweep and an audit row for no new reading. - if (insertedCount > 0 || updatedCount > 0) { - const silent = entries.length > PR_DETECTION_SILENT_THRESHOLD; - // v1.18.1 — eventful Vorsorge satisfaction. A matching reading just - // landed; resolve the user's reminders now rather than waiting on the - // 15-min cron. Fire-and-forget — the cron reconciles on enqueue miss. - fireAndForget(enqueueReminderSatisfy(user.id), { - action: "reminder.satisfy.enqueue", - }); - try { - await enqueuePrDetection(user.id, { silent }); - // Telemetry, not an audit event: it rides the wide event. - annotate({ - meta: { pr_detection_enqueued: true, pr_detection_silent: silent }, - }); - } catch (err) { - // Enqueue failure is non-fatal — the 30-minute fallback cron - // picks the user up in the next slot. Log so the operator - // notices repeated failures in Wide-Event traffic. - annotate({ - action: { name: "personal_records.detection_enqueue_failed" }, - meta: { error: err instanceof Error ? err.message : String(err) }, - }); - } - } - - annotate({ - action: { name: "measurement.batch.ingest" }, - meta: { - processed: entries.length, - inserted: insertedCount, - updated: updatedCount + unchangedStatsCount, - stats_unchanged: unchangedStatsCount, - duplicates: duplicateCount, - skipped: skipped.length, - // v1.32.1 (issue #585) — per-reason skip breakdown; see the - // `skippedByReason` build above. - skipped_by_reason: skippedByReason, - failed: failedCount, - // v1.32.8 (iOS #66) — diagnostic sync-trigger tag (foreground / - // background / push / manual), or null when the client did not send - // one. It never influenced any count above; since v1.42 it decides - // how hard the caches below are invalidated. - syncTrigger: syncTrigger ?? null, - // v1.42 — duplicates refused because a live `stats:` row already - // covers the sample's hour or day (`folded-window.ts`). - folded_window: foldedWindowCount, - // v1.44 — inserts that took the set-based path rather than one - // reconciler transaction each. - set_inserted: setInsertedCount, - }, - }); - - // v1.4.34 — bust per-user analytics + achievements + workouts - // caches when at least one row landed so the next read picks up the - // ingested batch. Skipped-only ingests don't change state, so we - // gate the invalidation on `insertedCount > 0 || updatedCount > 0` - // — overwriting a per-day `stats:*` row changes the day's reading - // and must invalidate every consumer that reads through this user's - // measurements. - if (insertedCount > 0 || updatedCount > 0) { - // v1.42 — evict only when someone is looking. v1.18.9 (#38) evicted on - // every batch so a focus-refetch after a sync showed post-sync data, but - // a phone syncing in the background posts a batch every few minutes all - // day, and each eviction sent the next dashboard read into a cold - // rebuild. Every other background sync marks stale, which `cachedSwr` - // serves once while one rebuild runs; this route now does the same - // unless the app is open (`foreground`), the person pressed sync - // (`manual`), or the batch wrote one of their own hand-entered readings - // (`MANUAL`), whose entry has to show on the very next read. - const wroteManualRow = prepared.some( - (p) => - p.row.source === "MANUAL" && - (results[p.index]?.status === "inserted" || - results[p.index]?.status === "updated"), - ); - const evict = - syncTrigger === "foreground" || - syncTrigger === "manual" || - wroteManualRow; - // Every row this batch created, changed or retired is in - // `writtenIdentities`, so its earliest `measuredAt` bounds what changed: - // closed blocks of older readings (the pre-fold remainder, the day - // view's past history blocks) stay cached across the sync. - const touchedFrom = new Date( - Math.min(...writtenIdentities.map((w) => w.measuredAt.getTime())), - ); - invalidateUserMeasurements(user.id, { - ...(evict ? { evict: true } : {}), - ...(Number.isFinite(touchedFrom.getTime()) ? { touchedFrom } : {}), - }); - - // v1.37.19 — shared post-mutation tail for every distinct - // (type, day) the batch touched. Updates participate as well as - // inserts, so this deliberately retains the prepared-row scope. - // A batch is the phone's sync, held to the sync gap, unless it wrote one - // of the person's own hand-entered readings, which is narrated like any - // other hand entry. - await afterMeasurementMutation(user.id, writtenIdentities, { - source: wroteManualRow ? "manual" : "sync", - }); - - // Apple-Health sleep is the third sleep transport. If any SLEEP_DURATION - // row for last night just landed, kick the debounced morning refresh so the - // digest/score finalise with the current sleep. The trigger judges "last - // night" in the user's profile tz, so a historical Apple export replaying - // months of nights never re-triggers for an old night. Fire-and-forget. - void maybeEnqueueMorningRefresh( - user.id, - freshlyInsertedRows - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt), - ).catch(() => {}); - - // v1.31.0 — the measurement arm of the data-arrival spine. - // - // ONE emit per kind per request, never one per row. The batch route is the - // Apple-Health / iOS ingest path, so it is also the single biggest backfill - // path in the product: a ten-year export arrives here in 500-row batches. - // Emitting per row would put thousands of classifier calls (and thousands - // of dropped-event annotations) on the hot path for no gain, since the - // day-scoped singleton key collapses them to one job anyway. - // - // Blood pressure is deliberately one arrival, not two: it is stored as two - // rows (SYS + DIA) sharing a `measuredAt`, and a reader cares about the - // reading, not the arms. - // - // Only rows the write actually INSERTED count — `updatedCount` rows are the - // `stats:` overwrite contract re-stating a value the record already had, - // which is not news. If nothing was inserted the classifier returns `noop` - // and nothing is enqueued. - for (const [kind, group] of groupRowsByArrivalKind(freshlyInsertedRows)) { - void emitDataArrival({ - userId: user.id, - kind, - newestSampleAt: group.newestAt, - insertedCount: group.count, - source: "batch", - }).catch(() => {}); - } - } - - // v1.43 — the hours are not measurements, so they touched none of the - // measurement consumers above; only the caches that read the intraday - // profile (the same-time baseline) are marked stale. - if ( - hourInsertedCount + hourUpdatedCount > 0 && - insertedCount === 0 && - updatedCount === 0 - ) { - invalidateUserMeasurements(user.id); - } - - const response = apiSuccess({ - processed: entries.length, - inserted: insertedCount + hourInsertedCount, - updated: updatedCount + unchangedStatsCount + hourUpdatedCount, - duplicates: duplicateCount, - failed: failedCount, - skipped, - entries: results, - }); - if (failedCount > 0) { + const response = apiSuccess(result); + if (result.failed > 0) { response.headers.set("Cache-Control", "private, no-store"); } return response; diff --git a/src/lib/measurements/batch-ingest/entries.ts b/src/lib/measurements/batch-ingest/entries.ts new file mode 100644 index 000000000..f247682e2 --- /dev/null +++ b/src/lib/measurements/batch-ingest/entries.ts @@ -0,0 +1,115 @@ +/** + * The vocabulary of the native batch ingest (`POST /api/measurements/batch`): + * what one posted entry carries, what the ingest answers for it, and the two + * small rules more than one stage needs. + * + * The wire schema itself stays in the route, which owns the contract; the + * entry type here is what that schema parses to, and the route assigns one to + * the other so the compiler holds them together. + */ +import type { Prisma } from "@/generated/prisma/client"; +import type { HealthKitRequestSyncTrigger } from "@/lib/integrations/healthkit-type-sync"; +import { getUnitForType, VALUE_RANGES } from "@/lib/validations/measurement"; +import type { DeviceType } from "@/lib/validations/source-priority"; + +/** One HealthKit sample (or `stats:` aggregate) as the route parsed it. */ +export interface BatchIngestEntry { + hkIdentifier: string; + value: number; + unit: string; + startDate: string; + endDate: string; + sleepStage?: number; + categoryValue?: number; + valueMin?: number; + valueMax?: number; + externalId: string; + externalSourceVersion?: string; + source?: "APPLE_HEALTH" | "MANUAL"; + deviceType?: DeviceType | null; +} + +/** The parsed request: the entries and the client's declared wake reason. */ +export interface BatchIngestInput { + user: { id: string; timezone: string | null }; + /** A narrow single-purpose Bearer (a bridge), not the native client. */ + scoped: boolean; + entries: BatchIngestEntry[]; + syncTrigger: HealthKitRequestSyncTrigger | undefined; + ipAddress: string | null; +} + +/** + * Per-entry outcome the iOS client uses to advance its sync cursor. + * `inserted` and `duplicate` both indicate the row landed (or was + * already present), so the client can checkpoint past them. `updated` + * is the per-day-aggregate overwrite path: when the iOS client re-posts + * a `stats:*` cumulative row (Steps, Active Energy, Sleep Duration, + * Walking/Running Distance, Flights Climbed — see issue #213) for a day + * we already have, the server overwrites the row's `value` rather than + * dropping the new payload as a duplicate. Sample-class rows (every + * other HK metric) keep the strict `duplicate` semantics because each + * sample is a canonical, immutable reading. `skipped` is a validation no-op; + * `failed` is a hard database verdict and must not be checkpointed as landed. + */ +export type EntryStatus = + "inserted" | "updated" | "duplicate" | "skipped" | "failed"; + +export interface EntryResult { + index: number; + status: EntryStatus; + reason?: string; + /** + * Only on a `value_out_of_range` skip: the value after conversion into the + * stored unit, which is what the band was checked against. + */ + convertedValue?: number; + /** Only on a `value_out_of_range` skip: the band applied, in the stored unit. */ + range?: { min: number; max: number; unit: string }; +} + +/** An entry that passed the first pass, with the row it becomes. */ +export interface PreparedEntry { + index: number; + entry: BatchIngestEntry; + row: Prisma.MeasurementCreateManyInput; +} + +// Issue #213 — `stats:*` externalIds carry per-day cumulative +// totals that the iOS HealthKit observer re-posts as the day progresses. +// Treating those as immutable duplicates freezes the day's tile at the +// first-sync value. Sample-class externalIds (every other prefix) keep +// the strict insert-only contract. +const STATS_EXTERNAL_ID_PREFIX = "stats:"; + +export function isStatsExternalId( + externalId: string | null | undefined, +): boolean { + return ( + typeof externalId === "string" && + externalId.startsWith(STATS_EXTERNAL_ID_PREFIX) + ); +} + +/** + * The `value_out_of_range` skip, with what a client needs to say which limit + * applied. The value is the entry's own reading after conversion, handed back + * to the account that sent it; a fraction sent where a percent was expected + * shows up here as a converted value a hundred times too small or too large. + */ +export function outOfRangeResult( + index: number, + type: string, + convertedValue: number, +): EntryResult { + const band = VALUE_RANGES[type]; + return { + index, + status: "skipped", + reason: "value_out_of_range", + ...(Number.isFinite(convertedValue) ? { convertedValue } : {}), + ...(band + ? { range: { min: band.min, max: band.max, unit: getUnitForType(type) } } + : {}), + }; +} diff --git a/src/lib/measurements/batch-ingest/ingest.ts b/src/lib/measurements/batch-ingest/ingest.ts new file mode 100644 index 000000000..38f22c6f3 --- /dev/null +++ b/src/lib/measurements/batch-ingest/ingest.ts @@ -0,0 +1,125 @@ +/** + * The native batch ingest behind `POST /api/measurements/batch`, from a + * parsed and admitted request to the per-entry answer. + * + * The stages run in a fixed order, and the order is part of the contract: + * the first pass maps every entry; the measurements are written, then the + * hourly sums; the sync checkpoint is stamped from the verdicts; the audit + * row, the wide-event breakdowns and the record jobs follow; the caches and + * the arrival spine hear last. The route owns authentication, the rate + * limit, the body and the response envelope. + */ +import { annotate } from "@/lib/logging/context"; +import { unstableExternalIdMeta } from "@/lib/validations/external-id"; + +import type { BatchIngestInput, EntryResult } from "./entries"; +import { prepareBatch } from "./prepare"; +import { + annotateIngest, + enqueueRecordJobs, + reportBatchOutcome, + summariseSkips, + type BatchTallies, + type SkippedEntry, +} from "./report"; +import { stampSyncCheckpoint } from "./sync-checkpoint"; +import { afterBatchWrite } from "./tail"; +import { writeBatchHours } from "./write-hours"; +import { MeasurementBatchWriter } from "./write-measurements"; + +/** The response body of the batch route. */ +export interface BatchIngestResult { + processed: number; + inserted: number; + updated: number; + duplicates: number; + failed: number; + skipped: SkippedEntry[]; + entries: EntryResult[]; +} + +export async function ingestMeasurementBatch( + input: BatchIngestInput, +): Promise { + const { user, scoped, entries, syncTrigger, ipAddress } = input; + + const batch = await prepareBatch(input); + const { prepared, results } = batch; + if (batch.unstableShapes.length > 0) { + annotate({ + meta: unstableExternalIdMeta("measurement.batch", batch.unstableShapes), + }); + } + + const writer = new MeasurementBatchWriter(user.id, results); + await writer.write(prepared); + + const hours = await writeBatchHours({ + userId: user.id, + hourWrites: batch.hourWrites, + accountTz: batch.accountTz, + results, + processed: entries.length, + }); + writer.duplicateCount += hours.duplicates; + writer.failedCount += hours.failed; + + await stampSyncCheckpoint({ + userId: user.id, + scoped, + syncTrigger, + prepared, + results, + failedCount: writer.failedCount, + unchangedStatsRows: writer.unchangedStatsRows, + hourLanded: hours.hourLanded, + hourChanged: hours.hourChanged, + }); + + const tallies: BatchTallies = writer; + const { skipped, skippedByReason } = summariseSkips(results); + await reportBatchOutcome({ + userId: user.id, + ipAddress, + entries, + results, + tallies, + skipped, + }); + await enqueueRecordJobs({ + userId: user.id, + processed: entries.length, + tallies, + }); + annotateIngest({ + processed: entries.length, + tallies, + skipped, + skippedByReason, + syncTrigger, + }); + + await afterBatchWrite({ + userId: user.id, + syncTrigger, + prepared, + results, + insertedCount: writer.insertedCount, + updatedCount: writer.updatedCount, + insertedPrepared: writer.insertedPrepared, + writtenIdentities: writer.writtenIdentities, + hourInsertedCount: hours.hourInsertedCount, + hourUpdatedCount: hours.hourUpdatedCount, + }); + + return { + processed: entries.length, + inserted: writer.insertedCount + hours.hourInsertedCount, + updated: + writer.updatedCount + writer.unchangedStatsCount + hours.hourUpdatedCount, + duplicates: writer.duplicateCount, + failed: writer.failedCount, + skipped, + entries: results, + }; +} diff --git a/src/lib/measurements/batch-ingest/prepare.ts b/src/lib/measurements/batch-ingest/prepare.ts new file mode 100644 index 000000000..fc5eb9d2b --- /dev/null +++ b/src/lib/measurements/batch-ingest/prepare.ts @@ -0,0 +1,310 @@ +/** + * First pass of the batch ingest: map each posted entry to the row it would + * become, or to the per-entry refusal that keeps it out. Nothing is written + * here; hourly sums are validated and set aside for their own write. + */ +import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source"; +import { + isAggregatedBucketExternalId, + mapAppleHealthEntry, + targetsAggregatedBucket, +} from "@/lib/measurements/apple-health-mapping"; +import { + classifyHourBucket, + HOUR_BUCKET_SKIP_REASONS, + parseHourBucketId, + type HourBucketEntry, + type HourBucketId, +} from "@/lib/measurements/intraday-hour-buckets"; +import { resolveStoredTimezone } from "@/lib/tz/resolver"; +import { isPlausibleEntryInstant } from "@/lib/validations/entry-instant"; +import { + classifyExternalId, + UNSTABLE_EXTERNAL_ID_REASON, + type UnstableExternalIdShape, +} from "@/lib/validations/external-id"; +import { validateMeasurementRange } from "@/lib/validations/measurement"; + +import { + outOfRangeResult, + type BatchIngestEntry, + type BatchIngestInput, + type EntryResult, + type PreparedEntry, +} from "./entries"; + +export interface PreparedBatch { + prepared: PreparedEntry[]; + results: EntryResult[]; + unstableShapes: UnstableExternalIdShape[]; + /** Hourly sums for the intraday profile, written after the measurements. */ + hourWrites: HourBucketEntry[]; + /** The zone the hours are cut in, resolved when the batch carried one. */ + accountTz: string | undefined; +} + +export async function prepareBatch( + input: Pick, +): Promise { + const { user, scoped, entries } = input; + const batch: PreparedBatch = { + prepared: [], + results: new Array(entries.length), + unstableShapes: [], + hourWrites: [], + accountTz: undefined, + }; + const batchNow = new Date(); + + for (let index = 0; index < entries.length; index++) { + const entry = entries[index]; + + // The `(user_id, type, source, external_id)` dedup key above is only + // idempotent while the id is STABLE across client launches. An id + // that rotates per process (an object description carrying a memory + // address) never matches its own earlier row, so every sync sweep + // mints a fresh measurement. Refused per entry, like the range and + // timestamp guards below — one malformed row must not stop the other + // 499 from landing. + const unstable = classifyExternalId(entry.externalId); + if (unstable) { + batch.unstableShapes.push(unstable); + batch.results[index] = { + index, + status: "skipped", + reason: UNSTABLE_EXTERNAL_ID_REASON, + }; + continue; + } + + // An hourly sum of a summed type (`stats::T`, ios #77) + // is never a measurement: it lands in the intraday profile behind the + // same-time baseline, and the day's total stays the day row. Validated + // here, written after the measurements. + const hourId = parseHourBucketId(entry.externalId); + if (hourId) { + await prepareHourEntry(batch, { + index, + entry, + hourId, + user, + scoped, + batchNow, + }); + continue; + } + + const prepared = prepareMeasurementEntry(index, entry, user.id, scoped); + if ("row" in prepared) batch.prepared.push(prepared); + else batch.results[index] = prepared; + } + return batch; +} + +/** Validate one hourly sum and set it aside, or record why it is refused. */ +async function prepareHourEntry( + batch: PreparedBatch, + args: { + index: number; + entry: BatchIngestEntry; + hourId: HourBucketId; + user: BatchIngestInput["user"]; + scoped: boolean; + batchNow: Date; + }, +): Promise { + const { index, entry, hourId, user, scoped, batchNow } = args; + const skip = (reason: string) => { + batch.results[index] = { index, status: "skipped", reason }; + }; + // The profile has no source dimension and is the phone's: a bridge or a + // hand-entered row cannot post hours into it. + if (scoped || (entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") { + skip(HOUR_BUCKET_SKIP_REASONS.unsupportedType); + return; + } + batch.accountTz ??= await resolveStoredTimezone(user.timezone); + const startDate = new Date(entry.startDate); + const verdict = classifyHourBucket( + hourId, + startDate, + batch.accountTz, + batchNow, + ); + if (!verdict.ok) { + skip(verdict.reason); + return; + } + if (!isPlausibleEntryInstant(startDate)) { + skip("implausible_timestamp"); + return; + } + const hourMapped = mapAppleHealthEntry({ + hkIdentifier: entry.hkIdentifier, + value: entry.value, + unit: entry.unit, + startDate: entry.startDate, + endDate: entry.endDate, + }); + if (!hourMapped || hourMapped.type !== verdict.type) { + // The entry's identifier must be the one its id names. + skip( + hourMapped + ? HOUR_BUCKET_SKIP_REASONS.unsupportedType + : "unmappable_identifier", + ); + return; + } + if ( + hourMapped.value < 0 || + validateMeasurementRange(hourMapped.type, hourMapped.value) !== null + ) { + batch.results[index] = outOfRangeResult( + index, + hourMapped.type, + hourMapped.value, + ); + return; + } + batch.hourWrites.push({ + index, + type: verdict.type, + hkIdentifier: hourId.hkIdentifier, + dateKey: hourId.dateKey, + hour: hourId.hour, + value: hourMapped.value, + startMs: startDate.getTime(), + }); +} + +/** Map one entry to its measurement row, or to the refusal that keeps it out. */ +function prepareMeasurementEntry( + index: number, + entry: BatchIngestEntry, + userId: string, + scoped: boolean, +): PreparedEntry | EntryResult { + const mapped = mapAppleHealthEntry({ + hkIdentifier: entry.hkIdentifier, + value: entry.value, + unit: entry.unit, + startDate: entry.startDate, + endDate: entry.endDate, + sleepStage: entry.sleepStage, + categoryValue: entry.categoryValue, + valueMin: entry.valueMin, + valueMax: entry.valueMax, + }); + + if (!mapped) { + return { + index, + status: "skipped", + reason: "unmappable_identifier", + }; + } + + // Plausibility-range guard. We tolerate "out of plausible range" + // by skipping rather than failing the batch — a single bogus + // sensor reading shouldn't poison an otherwise clean ingest. + const rangeError = validateMeasurementRange(mapped.type, mapped.value); + if (rangeError !== null) { + return outOfRangeResult(index, mapped.type, mapped.value); + } + + // Time-plausibility guard, the bound the single-entry twin already applies + // through `validateEntryInstant` on `createMeasurementSchema.measuredAt`. + // Without it a future-dated row permanently won "latest reading" on every + // dashboard and poisoned the rollup buckets and the canonical picker. The + // past floor stays at 1900 (no `maxAgeMs`) because a historical Apple + // Health backfill is a legitimate, expected shape on this route; only the + // future side is a bug. Skipped per entry, like the range guard above. + if (!isPlausibleEntryInstant(mapped.takenAt)) { + return { + index, + status: "skipped", + reason: "implausible_timestamp", + }; + } + + // The go-forward aggregated heart-rate wire contract (iOS #34). A + // `stats:HKQuantityTypeIdentifierHeartRate:<10-min-ISO-Z>` row is the + // 10-minute-average PULSE bucket; it rides the generic `stats:` overwrite + // path so a within-bucket re-post (the running mean shifts) replaces the + // value instead of duplicating. Reject a row that targets a bucket prefix + // but carries a malformed suffix — a garbage suffix would mint an + // un-overwriteable row the next re-post can't collapse onto. A well-formed + // bucket (and every non-bucket `stats:` / per-sample `uuid` externalId) + // passes through unchanged. + if ( + targetsAggregatedBucket(entry.externalId) && + !isAggregatedBucketExternalId(entry.externalId) + ) { + return { + index, + status: "skipped", + reason: "malformed_hr_bucket_id", + }; + } + + // The per-bucket MIN / MAX are persisted ONLY on a well-formed 10-min + // aggregated bucket row. A per-sample reading, a per-day cumulative + // `stats:` total, or a manual entry never carries a spread, so we pin them + // to null there even if a client mis-sends the fields — `value` stays the + // single source of truth for those rows. + // + // The spread is validated the same way `value` is guarded (above): it + // drives the DAY min/max band + the intraday envelope, so an out-of-range + // or mis-ordered spread (a spurious `discreteMin` of 0, a sensor glitch) + // would poison the aggregate exactly like a bogus `value` would. Persist + // the spread ONLY when both extremes are present, in range, and correctly + // ordered around `value`; otherwise drop to null and keep the trustworthy + // average. Partial or invalid spread → avg-only, still valid. + const isHrBucket = isAggregatedBucketExternalId(entry.externalId); + const spreadValid = + isHrBucket && + typeof mapped.valueMin === "number" && + typeof mapped.valueMax === "number" && + validateMeasurementRange(mapped.type, mapped.valueMin) === null && + validateMeasurementRange(mapped.type, mapped.valueMax) === null && + mapped.valueMin <= mapped.value && + mapped.value <= mapped.valueMax; + const valueMin = spreadValid ? mapped.valueMin : null; + const valueMax = spreadValid ? mapped.valueMax : null; + + return { + index, + entry, + row: { + userId, + type: mapped.type, + value: mapped.value, + valueMin, + valueMax, + unit: mapped.unit, + // Honour the per-entry source tag, defaulting to `APPLE_HEALTH` when + // absent so legacy callers are unchanged. + // + // A scoped credential gets `EXTERNAL`, and can hold no other value: the + // route refused any body that named one, so nothing asserted is being + // discarded here. The value has to move with the credential rather + // than with the route, because the route serves both the phone and a + // third-party bridge. It also moves the dedup key to + // `(userId, type, EXTERNAL, externalId)`, which is the isolated + // namespace this source exists to give. + source: scoped ? EXTERNAL_SOURCE : (entry.source ?? "APPLE_HEALTH"), + measuredAt: mapped.takenAt, + externalId: entry.externalId, + externalSourceVersion: entry.externalSourceVersion ?? null, + sleepStage: mapped.sleepStage ?? null, + // Categorical events: the device's own verdict / severity for an EVENT + // row; null for every continuous reading. + rhythmClassification: mapped.rhythmClassification ?? null, + // The iOS-supplied device-type, passed through to the row. Stays null + // for builds that do not send it; the canonical picker treats null as + // `unknown` so legacy ingest keeps working without a server-side + // default. + deviceType: entry.deviceType ?? null, + }, + }; +} diff --git a/src/lib/measurements/batch-ingest/report.ts b/src/lib/measurements/batch-ingest/report.ts new file mode 100644 index 000000000..8cce1851e --- /dev/null +++ b/src/lib/measurements/batch-ingest/report.ts @@ -0,0 +1,258 @@ +/** + * What a batch reports once its rows are written: the audit row, the + * operator's wide-event breakdowns, and the record-detection and reminder + * jobs a written row starts. + */ +import { fireAndForget } from "@/lib/logging/fire-and-forget"; +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import { targetsAggregatedBucket } from "@/lib/measurements/apple-health-mapping"; +import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; +import type { HealthKitRequestSyncTrigger } from "@/lib/integrations/healthkit-type-sync"; + +import type { BatchIngestEntry, EntryResult } from "./entries"; + +// Historical-backfill threshold for PR push suppression. A batch larger than +// this fires the detection job with `silent: true` so a multi-year Apple +// Health backfill writes records without spamming the user with hundreds of +// pushes. Tuned generously — a healthy daily-sync batch is typically well +// under 50 entries. +const PR_DETECTION_SILENT_THRESHOLD = 50; + +/** The measurement-side tallies the report reads. */ +export interface BatchTallies { + insertedCount: number; + updatedCount: number; + unchangedStatsCount: number; + duplicateCount: number; + failedCount: number; + crossSourceMergedCount: number; + foldedWindowCount: number; + setInsertedCount: number; +} + +export type SkippedEntry = { index: number; reason: string }; + +/** The skipped entries, and how many tripped each reason. */ +export function summariseSkips(results: EntryResult[]): { + skipped: SkippedEntry[]; + skippedByReason: Record; +} { + const skipped = results + .filter((r) => r.status === "skipped") + .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); + + // Per-reason skip breakdown (issue #585). The ingest annotation only ever + // carried the total `skipped` count, so an operator investigating "why does + // this account's uploaded data look sparser than expected" had no way to + // see WHICH validation gate an entry tripped (malformed bucket id, + // out-of-range value, unmappable identifier, …) without pulling raw rows. + // Grouping by reason turns that into a one-glance wide-event field. + const skippedByReason: Record = {}; + for (const s of skipped) { + skippedByReason[s.reason] = (skippedByReason[s.reason] ?? 0) + 1; + } + return { skipped, skippedByReason }; +} + +/** The audit row and the per-feature wide-event breakdowns. */ +export async function reportBatchOutcome(args: { + userId: string; + ipAddress: string | null; + entries: BatchIngestEntry[]; + results: EntryResult[]; + tallies: BatchTallies; + skipped: SkippedEntry[]; +}): Promise { + const { userId, ipAddress, entries, results, tallies, skipped } = args; + await auditLog("measurement.batch.ingest", { + userId, + ipAddress, + details: { + processed: entries.length, + inserted: tallies.insertedCount, + updated: tallies.updatedCount + tallies.unchangedStatsCount, + duplicates: tallies.duplicateCount, + skipped: skipped.length, + failed: tallies.failedCount, + }, + }); + + // Issue #213 — a dedicated wide-event annotation so an operator can grep + // `measurement.batch.stats-overwrite` to see how often per-day cumulative + // rows are getting re-posted (a healthy ingest flow). Only fires when at + // least one row was overwritten so the baseline ingest trace is unchanged + // for batches with no `stats:*` duplicates. + if (tallies.updatedCount > 0) { + annotate({ + action: { name: "measurement.batch.stats-overwrite" }, + meta: { + updated: tallies.updatedCount, + processed: entries.length, + }, + }); + } + + // iOS #2 — a dedicated annotation for the MANUAL↔APPLE_HEALTH same-reading + // collapse so an operator can grep `measurement.batch.cross-source-merge` + // to confirm the standalone-pair mirror duplicate is being absorbed (a + // healthy first-pair flow). Only fires when at least one row was merged so + // the baseline ingest trace is unchanged for the common case. + if (tallies.crossSourceMergedCount > 0) { + annotate({ + action: { name: "measurement.batch.cross-source-merge" }, + meta: { + merged: tallies.crossSourceMergedCount, + processed: entries.length, + }, + }); + } + + annotateHeartRateBuckets(entries, results); +} + +/** + * Issue #585 — a dedicated outcome breakdown for entries targeting the + * go-forward aggregated 10-min heart-rate bucket prefix + * (`stats:HKQuantityTypeIdentifierHeartRate:<10-min-Z>`). The intraday pulse + * chart reads exactly these rows for a live-synced day; a report of "the + * chart looks sparser after the ZIP import cutoff" is unanswerable from the + * baseline ingest trace alone, since it never distinguished an HR-bucket + * write from any other entry in the batch. Grepping + * `measurement.batch.hr-bucket` shows, per sync, how many bucket entries the + * client sent and what happened to each — confirming (or ruling out) + * server-side rejection/dedup as the density-gap cause. Only fires when the + * batch actually carried at least one HR-bucket entry so the baseline trace + * is unchanged for every other sync. + */ +function annotateHeartRateBuckets( + entries: BatchIngestEntry[], + results: EntryResult[], +): void { + let hrBucketInserted = 0; + let hrBucketUpdated = 0; + let hrBucketDuplicate = 0; + let hrBucketSkipped = 0; + let hrBucketFailed = 0; + for (let index = 0; index < entries.length; index++) { + if (!targetsAggregatedBucket(entries[index].externalId)) continue; + switch (results[index]?.status) { + case "inserted": + hrBucketInserted++; + break; + case "updated": + hrBucketUpdated++; + break; + case "duplicate": + hrBucketDuplicate++; + break; + case "skipped": + hrBucketSkipped++; + break; + case "failed": + hrBucketFailed++; + break; + } + } + const hrBucketAttempted = + hrBucketInserted + + hrBucketUpdated + + hrBucketDuplicate + + hrBucketSkipped + + hrBucketFailed; + if (hrBucketAttempted > 0) { + annotate({ + action: { name: "measurement.batch.hr-bucket" }, + meta: { + attempted: hrBucketAttempted, + inserted: hrBucketInserted, + updated: hrBucketUpdated, + duplicate: hrBucketDuplicate, + skipped: hrBucketSkipped, + failed: hrBucketFailed, + processed: entries.length, + }, + }); + } +} + +/** + * Kick off PR detection for this user when at least one row was written, so + * a single off-day reading still gets evaluated; the warm-up gate inside the + * detector decides whether it's a record. Suppress push notifications for + * historical backfills above the silent threshold. Updated `stats:*` rows + * also count — a per-day-total overwrite can flip the day's value past a + * personal record. + * + * A batch of duplicates only does not enqueue. It changed nothing a record + * or a reminder could read, and an anchor-reset backfill sends hundreds of + * them, each of which used to queue a detection pass, a reminder sweep and + * an audit row for no new reading. + */ +export async function enqueueRecordJobs(args: { + userId: string; + processed: number; + tallies: Pick; +}): Promise { + const { userId, processed, tallies } = args; + if (!(tallies.insertedCount > 0 || tallies.updatedCount > 0)) return; + const silent = processed > PR_DETECTION_SILENT_THRESHOLD; + // Eventful Vorsorge satisfaction. A matching reading just landed; resolve + // the user's reminders now rather than waiting on the 15-min cron. + // Fire-and-forget — the cron reconciles on enqueue miss. + fireAndForget(enqueueReminderSatisfy(userId), { + action: "reminder.satisfy.enqueue", + }); + try { + await enqueuePrDetection(userId, { silent }); + // Telemetry, not an audit event: it rides the wide event. + annotate({ + meta: { pr_detection_enqueued: true, pr_detection_silent: silent }, + }); + } catch (err) { + // Enqueue failure is non-fatal — the 30-minute fallback cron + // picks the user up in the next slot. Log so the operator + // notices repeated failures in Wide-Event traffic. + annotate({ + action: { name: "personal_records.detection_enqueue_failed" }, + meta: { error: err instanceof Error ? err.message : String(err) }, + }); + } +} + +/** The batch's own wide-event summary, the last annotation of the request. */ +export function annotateIngest(args: { + processed: number; + tallies: BatchTallies; + skipped: SkippedEntry[]; + skippedByReason: Record; + syncTrigger: HealthKitRequestSyncTrigger | undefined; +}): void { + const { processed, tallies, skipped, skippedByReason, syncTrigger } = args; + annotate({ + action: { name: "measurement.batch.ingest" }, + meta: { + processed, + inserted: tallies.insertedCount, + updated: tallies.updatedCount + tallies.unchangedStatsCount, + stats_unchanged: tallies.unchangedStatsCount, + duplicates: tallies.duplicateCount, + skipped: skipped.length, + // Per-reason skip breakdown (issue #585); see `summariseSkips`. + skipped_by_reason: skippedByReason, + failed: tallies.failedCount, + // Diagnostic sync-trigger tag (iOS #66: foreground / background / push + // / manual), or null when the client did not send one. It never + // influences any count above; it decides how hard the caches are + // invalidated. + syncTrigger: syncTrigger ?? null, + // Duplicates refused because a live `stats:` row already covers the + // sample's hour or day (`folded-window.ts`). + folded_window: tallies.foldedWindowCount, + // Inserts that took the set-based path rather than one reconciler + // transaction each. + set_inserted: tallies.setInsertedCount, + }, + }); +} diff --git a/src/lib/measurements/batch-ingest/sync-checkpoint.ts b/src/lib/measurements/batch-ingest/sync-checkpoint.ts new file mode 100644 index 000000000..b9203b71d --- /dev/null +++ b/src/lib/measurements/batch-ingest/sync-checkpoint.ts @@ -0,0 +1,132 @@ +/** + * The sync checkpoint and the HealthKit stamps a batch sets on the account. + * + * This batch route is the native ingestion boundary, so it is also where the + * SyncMode checkpoint (`User.lastSyncedAt`, the same column + * `markSyncCheckpoint` writes) is established — `GET /api/sync/state` only + * reports it. The checkpoint covers MANUAL pushes too: the client synced + * whatever it was holding. A batch carrying a hard failure leaves the + * checkpoint where it was, so the undelivered window stays inside whatever + * the client asks for next. + * + * The HealthKit fields ride the same update and keep their narrower + * condition. Stamp only after every attempted write has reached a durable, + * non-failed verdict; MANUAL-only and skipped-only batches are not + * HealthKit syncs. + * + * The trigger the client declared is stamped alongside, and a background- or + * push-triggered batch additionally dates `healthKitLastBackgroundSyncAt`. + * Without that second timestamp there is no way to tell a phone that delivers + * on its own from one that only ever delivers while the app is open — both + * look like a moving `healthKitLastSyncedAt`. A batch that declares no + * trigger writes null over the previous value rather than leaving a stale one + * standing: the field describes THIS sync, and an older build's silence is + * not evidence about it. + * + * None of that is true of a scoped credential, which is why it is excluded + * from the whole block rather than only from the HealthKit half. The + * checkpoint's meaning is "the client synced what it was holding", and a + * third-party bridge is not the client and holds nothing on its behalf. + * `lastSyncedAt` is what `GET /api/sync/state` reports and what the phone + * uses to size its undelivered window, so letting a scale advance it can make + * the phone skip a window of its own samples — a silent data loss whose cause + * would be nowhere near the code that caused it. + * + * A `backfill` batch (ios #77) is a catch-up, not a sync. It records its + * types under `lastBackfillAt` only and moves none of the live stamps: + * thirty days of history must not read as thirty healthy daily syncs. + */ +import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; +import { + BACKFILL_SYNC_TRIGGER, + recordHealthKitTypeArrivals, + tallyHealthKitArrivals, + type HealthKitRequestSyncTrigger, +} from "@/lib/integrations/healthkit-type-sync"; +import type { HourBucketWrite } from "@/lib/measurements/intraday-hour-buckets"; + +import type { EntryResult, PreparedEntry } from "./entries"; + +export async function stampSyncCheckpoint(args: { + userId: string; + scoped: boolean; + syncTrigger: HealthKitRequestSyncTrigger | undefined; + prepared: PreparedEntry[]; + results: EntryResult[]; + failedCount: number; + unchangedStatsRows: Set; + hourLanded: HourBucketWrite[]; + hourChanged: Set; +}): Promise { + const { + userId, + scoped, + syncTrigger, + prepared, + results, + failedCount, + unchangedStatsRows, + hourLanded, + hourChanged, + } = args; + + const healthKitSyncSucceeded = + failedCount === 0 && + (hourLanded.length > 0 || + prepared.some((p) => { + if ((p.entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") return false; + const status = results[p.index]?.status; + return ( + status === "inserted" || + status === "updated" || + status === "duplicate" + ); + })); + + const backfill = syncTrigger === BACKFILL_SYNC_TRIGGER; + if (failedCount === 0 && !scoped) { + const syncedAt = new Date(); + const deliveredWithoutTheApp = + syncTrigger === "background" || syncTrigger === "push"; + // Per-type arrivals for the Apple Health card (#1173), Apple Health rows + // and hours only, under the checkpoint's conditions. Since #1259 only a + // new or changed one brings something (`brought`). + const arrivals = tallyHealthKitArrivals([ + ...prepared.flatMap((p) => { + const status = results[p.index]?.status; + if (p.row.source !== "APPLE_HEALTH" || !status) return []; + if (status === "skipped" || status === "failed") return []; + const unchanged = unchangedStatsRows.has(p); + const brought = status !== "duplicate" && !unchanged; + return [{ type: p.row.type as MeasurementType, brought }]; + }), + ...hourLanded.map((w) => ({ type: w.type, brought: hourChanged.has(w) })), + ]); + if (arrivals.size > 0) { + await recordHealthKitTypeArrivals(prisma, { + userId, + counts: arrivals, + trigger: syncTrigger ?? null, + at: syncedAt, + }); + } + if (!backfill) { + await prisma.user.update({ + where: { id: userId }, + data: { + lastSyncedAt: syncedAt, + ...(healthKitSyncSucceeded + ? { + healthKitLastSyncedAt: syncedAt, + healthKitLastSyncTrigger: syncTrigger ?? null, + ...(deliveredWithoutTheApp + ? { healthKitLastBackgroundSyncAt: syncedAt } + : {}), + } + : {}), + }, + }); + } + } +} diff --git a/src/lib/measurements/batch-ingest/tail.ts b/src/lib/measurements/batch-ingest/tail.ts new file mode 100644 index 000000000..19dbaf7e4 --- /dev/null +++ b/src/lib/measurements/batch-ingest/tail.ts @@ -0,0 +1,152 @@ +/** + * The consumers a written batch has to tell: the measurement caches, the + * rollup and status tail, the morning refresh and the data-arrival spine. + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; +import { emitDataArrival } from "@/lib/arrivals/emit-shared"; +import { groupRowsByArrivalKind } from "@/lib/arrivals/measurement-kind"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import type { HealthKitRequestSyncTrigger } from "@/lib/integrations/healthkit-type-sync"; + +import type { EntryResult, PreparedEntry } from "./entries"; + +type Identity = { type: MeasurementType; measuredAt: Date }; + +export async function afterBatchWrite(args: { + userId: string; + syncTrigger: HealthKitRequestSyncTrigger | undefined; + prepared: PreparedEntry[]; + results: EntryResult[]; + insertedCount: number; + updatedCount: number; + insertedPrepared: PreparedEntry[]; + writtenIdentities: Identity[]; + hourInsertedCount: number; + hourUpdatedCount: number; +}): Promise { + const { userId, insertedCount, updatedCount } = args; + const freshlyInsertedRows = args.insertedPrepared.map((p) => ({ + type: p.row.type as MeasurementType, + measuredAt: + p.row.measuredAt instanceof Date + ? p.row.measuredAt + : new Date(p.row.measuredAt as string), + })); + + // Bust per-user analytics + achievements + workouts caches when at least + // one row landed so the next read picks up the ingested batch. Skipped-only + // ingests don't change state, so the invalidation is gated on + // `insertedCount > 0 || updatedCount > 0` — overwriting a per-day `stats:*` + // row changes the day's reading and must invalidate every consumer that + // reads through this user's measurements. + if (insertedCount > 0 || updatedCount > 0) { + await afterMeasurementsWritten({ ...args, freshlyInsertedRows }); + } + + // The hours are not measurements, so they touched none of the measurement + // consumers above; only the caches that read the intraday profile (the + // same-time baseline) are marked stale. + if ( + args.hourInsertedCount + args.hourUpdatedCount > 0 && + insertedCount === 0 && + updatedCount === 0 + ) { + invalidateUserMeasurements(userId); + } +} + +async function afterMeasurementsWritten(args: { + userId: string; + syncTrigger: HealthKitRequestSyncTrigger | undefined; + prepared: PreparedEntry[]; + results: EntryResult[]; + writtenIdentities: Identity[]; + freshlyInsertedRows: Identity[]; +}): Promise { + const { + userId, + syncTrigger, + prepared, + results, + writtenIdentities, + freshlyInsertedRows, + } = args; + // Evict only when someone is looking (#38 evicted on every batch so a + // focus-refetch after a sync showed post-sync data, but a phone syncing in + // the background posts a batch every few minutes all day, and each + // eviction sent the next dashboard read into a cold rebuild). Every other + // background sync marks stale, which `cachedSwr` serves once while one + // rebuild runs; this route does the same unless the app is open + // (`foreground`), the person pressed sync (`manual`), or the batch wrote + // one of their own hand-entered readings (`MANUAL`), whose entry has to + // show on the very next read. + const wroteManualRow = prepared.some( + (p) => + p.row.source === "MANUAL" && + (results[p.index]?.status === "inserted" || + results[p.index]?.status === "updated"), + ); + const evict = + syncTrigger === "foreground" || syncTrigger === "manual" || wroteManualRow; + // Every row this batch created, changed or retired is in + // `writtenIdentities`, so its earliest `measuredAt` bounds what changed: + // closed blocks of older readings (the pre-fold remainder, the day + // view's past history blocks) stay cached across the sync. + const touchedFrom = new Date( + Math.min(...writtenIdentities.map((w) => w.measuredAt.getTime())), + ); + invalidateUserMeasurements(userId, { + ...(evict ? { evict: true } : {}), + ...(Number.isFinite(touchedFrom.getTime()) ? { touchedFrom } : {}), + }); + + // The shared post-mutation tail for every distinct (type, day) the batch + // touched. Updates participate as well as inserts, so this deliberately + // retains the prepared-row scope. A batch is the phone's sync, held to the + // sync gap, unless it wrote one of the person's own hand-entered readings, + // which is narrated like any other hand entry. + await afterMeasurementMutation(userId, writtenIdentities, { + source: wroteManualRow ? "manual" : "sync", + }); + + // Apple-Health sleep is the third sleep transport. If any SLEEP_DURATION + // row for last night just landed, kick the debounced morning refresh so the + // digest/score finalise with the current sleep. The trigger judges "last + // night" in the user's profile tz, so a historical Apple export replaying + // months of nights never re-triggers for an old night. Fire-and-forget. + void maybeEnqueueMorningRefresh( + userId, + freshlyInsertedRows + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt), + ).catch(() => {}); + + // The measurement arm of the data-arrival spine. + // + // ONE emit per kind per request, never one per row. The batch route is the + // Apple-Health / iOS ingest path, so it is also the single biggest backfill + // path in the product: a ten-year export arrives here in 500-row batches. + // Emitting per row would put thousands of classifier calls (and thousands + // of dropped-event annotations) on the hot path for no gain, since the + // day-scoped singleton key collapses them to one job anyway. + // + // Blood pressure is deliberately one arrival, not two: it is stored as two + // rows (SYS + DIA) sharing a `measuredAt`, and a reader cares about the + // reading, not the arms. + // + // Only rows the write actually INSERTED count — updated rows are the + // `stats:` overwrite contract re-stating a value the record already had, + // which is not news. If nothing was inserted the classifier returns `noop` + // and nothing is enqueued. + for (const [kind, group] of groupRowsByArrivalKind(freshlyInsertedRows)) { + void emitDataArrival({ + userId, + kind, + newestSampleAt: group.newestAt, + insertedCount: group.count, + source: "batch", + }).catch(() => {}); + } +} diff --git a/src/lib/measurements/batch-ingest/write-hours.ts b/src/lib/measurements/batch-ingest/write-hours.ts new file mode 100644 index 000000000..8d5383daf --- /dev/null +++ b/src/lib/measurements/batch-ingest/write-hours.ts @@ -0,0 +1,99 @@ +/** + * The hourly sums of a batch, written to the intraday profile after the + * measurements. One write per slot: the last entry for one start wins (an + * earlier one in the same batch is a superseded duplicate), and the two + * starts of the hour a fall-back day repeats are summed + * (`combineHourWrites`). + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { + combineHourWrites, + writeHourBuckets, + type HourBucketEntry, + type HourBucketWrite, +} from "@/lib/measurements/intraday-hour-buckets"; + +import type { EntryResult } from "./entries"; + +export interface HourOutcome { + hourInsertedCount: number; + hourUpdatedCount: number; + hourLanded: HourBucketWrite[]; + /** #1259: the slots whose value changed, as opposed to an unchanged re-post. */ + hourChanged: Set; + /** Superseded entries, counted with the batch's duplicates. */ + duplicates: number; + /** Entries whose write failed, counted with the batch's failures. */ + failed: number; +} + +export async function writeBatchHours(args: { + userId: string; + hourWrites: HourBucketEntry[]; + accountTz: string | undefined; + results: EntryResult[]; + processed: number; +}): Promise { + const { userId, hourWrites, accountTz, results, processed } = args; + const outcome: HourOutcome = { + hourInsertedCount: 0, + hourUpdatedCount: 0, + hourLanded: [], + hourChanged: new Set(), + duplicates: 0, + failed: 0, + }; + if (!(hourWrites.length > 0 && accountTz)) return outcome; + + const { writes, superseded } = combineHourWrites(hourWrites); + for (const index of superseded) { + results[index] = { + index, + status: "duplicate", + reason: "superseded_in_batch", + }; + outcome.duplicates++; + } + const toWrite = writes.map(({ write }) => write); + try { + const statuses = await writeHourBuckets(prisma, userId, accountTz, toWrite); + for (const { write, contributors } of writes) { + const status = statuses.get(write.index) ?? "inserted"; + // An unchanged hour still reads `updated` on the wire (#1259). + const wire = status === "inserted" ? "inserted" : "updated"; + for (const index of contributors) { + results[index] = { index, status: wire }; + if (status === "inserted") outcome.hourInsertedCount++; + else outcome.hourUpdatedCount++; + } + outcome.hourLanded.push(write); + if (status !== "unchanged") outcome.hourChanged.add(write); + } + } catch (err) { + for (const { contributors } of writes) { + for (const index of contributors) { + results[index] = { + index, + status: "failed", + reason: "write_failed", + }; + outcome.failed++; + } + } + annotate({ + action: { name: "measurement.batch.hour-bucket.failed" }, + meta: { error: err instanceof Error ? err.message : String(err) }, + }); + } + annotate({ + action: { name: "measurement.batch.hour-bucket" }, + meta: { + attempted: hourWrites.length, + inserted: outcome.hourInsertedCount, + updated: outcome.hourUpdatedCount, + processed, + }, + }); + return outcome; +} diff --git a/src/lib/measurements/batch-ingest/write-measurements.ts b/src/lib/measurements/batch-ingest/write-measurements.ts new file mode 100644 index 000000000..225e78587 --- /dev/null +++ b/src/lib/measurements/batch-ingest/write-measurements.ts @@ -0,0 +1,566 @@ +/** + * Second pass of the batch ingest: write the prepared rows and record each + * one's verdict. + * + * Every row ends in one of the reconciler's verdicts, reached one of four + * ways: superseded by a later `stats:` entry in the same batch, an unchanged + * `stats:` re-post answered without a write, a duplicate found without a + * transaction (the cross-source merge, the exact stored duplicate, the + * folded window), or a write: the set-based insert for rows that share no + * identity with another entry, the per-entry reconciler for the rest. + */ +import { prisma } from "@/lib/db"; +import { + MEASURED_AT_TOLERANCE_MS, + isMergeableSource, + isSameReadingAcrossSource, + oppositeMergeSource, + type MergeCandidate, +} from "@/lib/measurements/cross-source-merge"; +import { + externalKeyOf, + onceInBatch, + readStoredSampleIdentities, + readStoredStatsRows, + sampleIsStoredDuplicate, + statsWriteUnchanged, +} from "@/lib/measurements/batch-stored-rows"; +import { + lockMeasurementUser, + reconcileExternalMeasurement, + type MeasurementReconciliationVerdict, +} from "@/lib/measurements/reconcile-external-measurement"; +import { classifyRangeRejection } from "@/lib/measurements/plausibility-gate"; +import { + FOLDED_WINDOW_REASON, + findFoldedWindowDuplicates, +} from "@/lib/measurements/folded-window"; +import { annotate } from "@/lib/logging/context"; +import type { + MeasurementSource, + MeasurementType, + Prisma, +} from "@/generated/prisma/client"; + +import { + isStatsExternalId, + outOfRangeResult, + type EntryResult, + type PreparedEntry, +} from "./entries"; + +/** + * Rows per set-based insert statement. 500 entries fit two statements; each + * row binds about fifteen parameters, far under PostgreSQL's 65 535. + */ +const SET_INSERT_CHUNK = 250; + +type Verdict = Pick & + Partial<{ + dirtyIdentities: Array<{ type: MeasurementType; measuredAt: Date }>; + error: { code?: string }; + }>; + +/** What the reads before the write loop found for this batch. */ +interface StoredState { + crossSourceCandidates: MergeCandidate[]; + onceInThisBatch: ReturnType; + existingByExternalKey: Awaited>; + storedStats: Awaited>; + foldedPrepared: Set; +} + +/** The reconciler's input for one prepared row. */ +function desiredFor(userId: string, p: PreparedEntry) { + const statsRow = isStatsExternalId(p.row.externalId as string); + return { + userId, + type: p.row.type as MeasurementType, + value: p.row.value as number, + valueMin: p.row.valueMin as number | null, + valueMax: p.row.valueMax as number | null, + unit: p.row.unit as string, + source: p.row.source as Prisma.MeasurementUncheckedCreateInput["source"], + measuredAt: p.row.measuredAt as Date, + externalId: p.row.externalId as string, + externalSourceVersion: p.row.externalSourceVersion as string | null, + glucoseContext: p.row + .glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"], + sleepStage: p.row + .sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"], + rhythmClassification: p.row + .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"], + deviceType: p.row.deviceType as string | null, + aggregationProvenance: statsRow ? ("HEALTHKIT_STATISTICS" as const) : null, + }; +} + +/** + * The reads the verdicts need, in the order the route has always made them: + * the cross-source candidates, the stored sample identities, the stored + * `stats:` rows, the folded windows. + */ +async function readStoredState( + userId: string, + prepared: PreparedEntry[], +): Promise { + // Same-reading merge for the MANUAL ↔ APPLE_HEALTH mirror duplicate + // (iOS #2). When the iOS app logs a manual reading offline it + // mirrors that reading into HealthKit; on pairing, adopt-on-pair + // uploads it as MANUAL and HealthKit background sync independently + // re-ingests the mirrored sample as APPLE_HEALTH. The two carry + // different externalIds + sources, so the composite unique index lets + // both land — duplicating the whole hand-logged history on first pair. + // We collapse them at ingest: an incoming MANUAL / APPLE_HEALTH row is + // dropped as a `duplicate` when a same-reading row of the OPPOSITE + // source already exists (same type + value + measuredAt within a tight + // ±2 s window). `stats:*` cumulative rows are excluded — those are + // per-day aggregates the observer overwrites in place, never + // hand-entered point readings. See `cross-source-merge.ts` for the + // full rule + why first-physical-reading-wins (the value is identical + // between the two rows, so only the source label differs). + // + // Pull the opposite-source candidate rows in one widened query: for + // every mergeable, non-`stats:*` incoming row, look for an opposite- + // source row of the same type whose measuredAt falls inside the + // tolerance window. Compared in JS because the value match needs a + // float epsilon the SQL `IN` can't express. + const mergeProbes = prepared.filter( + (p) => + isMergeableSource(p.row.source as string) && + !isStatsExternalId(p.row.externalId as string), + ); + const crossSourceCandidates: MergeCandidate[] = []; + if (mergeProbes.length > 0) { + const candidateRows = await prisma.measurement.findMany({ + where: { + userId, + deletedAt: null, + OR: mergeProbes.map((p) => { + const measuredAt = p.row.measuredAt as Date; + return { + type: p.row.type as MeasurementType, + source: oppositeMergeSource( + p.row.source as "MANUAL" | "APPLE_HEALTH", + ), + measuredAt: { + gte: new Date(measuredAt.getTime() - MEASURED_AT_TOLERANCE_MS), + lte: new Date(measuredAt.getTime() + MEASURED_AT_TOLERANCE_MS), + }, + }; + }), + }, + select: { type: true, source: true, value: true, measuredAt: true }, + }); + crossSourceCandidates.push(...candidateRows); + } + + // Exact-duplicate prefilter. A re-sent sample used to cost one + // interactive transaction of about ten round-trips (savepoint, three + // advisory locks, two candidate reads, a row lock) only for the + // reconciler to answer `duplicate` without writing. A backfill after an + // anchor reset re-sends whole 500-entry batches of exactly that. + // + // The reconciler answers `duplicate` for a non-`stats:` row exactly when + // an existing row, live OR tombstoned, has the same + // `(type, source, externalId)` AND the same `measuredAt` and + // `sleepStage`: the natural-identity index is unique NULLS NOT DISTINCT, + // so that row is both its external and its natural match, and with no + // aggregation provenance on a sample nothing else can win. One read over + // the batch's sample ids, without a `deletedAt` filter, answers that for + // every row at once. + // + // The verdict is taken at the very point the reconciler would have been + // called, after `superseded_in_batch` and the cross-source merge, so + // statuses, reasons, counters and the in-batch merge candidates are the + // same as before. A row is only prefiltered when neither its external + // key nor its natural identity occurs a second time in the batch: an + // earlier entry sharing either could move or retire the matched row + // before this one is reached, and then the reconciler has to decide. + // `stats:` rows are never prefiltered (overwrite semantics), and neither + // is a value the reconciler's own plausibility gate would refuse. + const onceInThisBatch = onceInBatch(prepared.map((p) => p.row)); + const existingByExternalKey = await readStoredSampleIdentities( + prisma, + userId, + [ + ...new Set( + prepared + .map((p) => p.row.externalId as string) + .filter((externalId) => !isStatsExternalId(externalId)), + ), + ], + ); + + // An unchanged `stats:` re-post is reported `updated` without a write + // (see `batch-stored-rows.ts`), under the same once-per-batch key rule. + const storedStats = await readStoredStatsRows(prisma, userId, [ + ...new Set( + prepared + .map((p) => p.row.externalId as string) + .filter((externalId) => isStatsExternalId(externalId)), + ), + ]); + + // The `folded_window` guard (`folded-window.ts`). A raw Apple Health + // sample of a dense or mean type, old enough to have been folded, whose + // hour or day a live `stats:` row already covers, is a duplicate: the fold + // no longer leaves the raw rows behind as tombstones, so this is what + // keeps a re-upload after an anchor reset from standing beside the mean + // that already accounts for it. One read for the batch, and none at all + // when no row is old enough. + const foldedIndexes = await findFoldedWindowDuplicates( + prisma, + userId, + prepared.map((p) => ({ + type: p.row.type as MeasurementType, + source: p.row.source as MeasurementSource, + externalId: p.row.externalId as string, + measuredAt: p.row.measuredAt as Date, + })), + ); + const foldedPrepared = new Set( + [...foldedIndexes].map((index) => prepared[index]), + ); + + return { + crossSourceCandidates, + onceInThisBatch, + existingByExternalKey, + storedStats, + foldedPrepared, + }; +} + +/** The measurement half of one batch: its verdicts and what they started. */ +export class MeasurementBatchWriter { + insertedCount = 0; + updatedCount = 0; + /** + * `stats:` re-posts reported `updated` with nothing to write (see + * `isUnchangedStatsRow`). Counted in the response's `updated` like before, + * and in nothing that starts work. + */ + unchangedStatsCount = 0; + readonly unchangedStatsRows = new Set(); // #1259: not an arrival + duplicateCount = 0; + failedCount = 0; + /** Subset of `insertedCount` written by the set-based insert. */ + setInsertedCount = 0; + /** + * Subset of `duplicateCount` collapsed by the MANUAL↔APPLE_HEALTH + * same-reading merge (iOS #2), as opposed to a plain composite-key + * duplicate. Surfaced as a dedicated wide-event count so an operator can + * see how often the standalone-pair mirror collapses. + */ + crossSourceMergedCount = 0; + /** Subset of `duplicateCount` refused by the `folded_window` guard. */ + foldedWindowCount = 0; + /** Only rows the reconciler actually inserted drive arrival side effects. */ + readonly insertedPrepared: PreparedEntry[] = []; + readonly writtenIdentities: Array<{ + type: MeasurementType; + measuredAt: Date; + }> = []; + + /** + * Rows already chosen for insert earlier in THIS batch — so a MANUAL + * and an APPLE_HEALTH same-reading arriving in the SAME batch also + * collapse (the DB probe only sees rows from prior batches). + */ + private readonly inBatchCandidates: MergeCandidate[] = []; + private readonly pendingInserts: PreparedEntry[] = []; + private stored!: StoredState; + + constructor( + private readonly userId: string, + private readonly results: EntryResult[], + ) {} + + /** Write every prepared row, in order, recording each verdict. */ + async write(prepared: PreparedEntry[]): Promise { + if (prepared.length === 0) return; + this.stored = await readStoredState(this.userId, prepared); + + // `stats:` rows carry overwrite semantics and the last entry in one batch + // is authoritative. Earlier snapshots are safe duplicates. + const lastStatsIndexByKey = new Map(); + for (const p of prepared) { + if (!isStatsExternalId(p.row.externalId as string)) continue; + lastStatsIndexByKey.set( + `${p.row.type}::${p.row.source}::${p.row.externalId}`, + p.index, + ); + } + + for (const p of prepared) { + await this.writeRow(p, lastStatsIndexByKey); + } + await this.flushPendingInserts(); + } + + private async writeRow( + p: PreparedEntry, + lastStatsIndexByKey: Map, + ): Promise { + const externalId = p.row.externalId as string; + const statsRow = isStatsExternalId(externalId); + const key = `${p.row.type}::${p.row.source}::${externalId}`; + if (statsRow && lastStatsIndexByKey.get(key) !== p.index) { + this.results[p.index] = { + index: p.index, + status: "duplicate", + reason: "superseded_in_batch", + }; + this.duplicateCount++; + return; + } + + if (statsRow && this.isUnchangedStatsRow(p)) { + this.results[p.index] = { index: p.index, status: "updated" }; + this.unchangedStatsCount++; + this.unchangedStatsRows.add(p); + this.pushMergeCandidate(p); + return; + } + + if (!statsRow && this.hasSameReadingSibling(p)) { + this.results[p.index] = { + index: p.index, + status: "duplicate", + reason: "cross_source_merge", + }; + this.duplicateCount++; + this.crossSourceMergedCount++; + return; + } + + // Both verdicts below are the reconciler's `duplicate`, reached without + // its transaction; like every duplicate they fall through to the + // in-batch merge candidates. + const prefiltered = !statsRow && this.isExactStoredDuplicate(p); + const folded = + !prefiltered && !statsRow && this.stored.foldedPrepared.has(p); + if (prefiltered || folded) { + this.results[p.index] = folded + ? { + index: p.index, + status: "duplicate", + reason: FOLDED_WINDOW_REASON, + } + : { index: p.index, status: "duplicate" }; + this.duplicateCount++; + if (folded) this.foldedWindowCount++; + this.pushMergeCandidate(p); + return; + } + + if (this.stored.onceInThisBatch(p.row)) { + this.pendingInserts.push(p); + this.pushMergeCandidate(p); + return; + } + + await this.flushPendingInserts(); + this.applyVerdict(p, await this.reconcileRow(p), true); + } + + private hasSameReadingSibling(p: PreparedEntry): boolean { + const incoming = { + type: p.row.type as MeasurementType, + source: p.row.source as string, + value: p.row.value as number, + measuredAt: p.row.measuredAt as Date, + }; + if (!isMergeableSource(incoming.source)) return false; + for (const candidate of this.stored.crossSourceCandidates) { + if (isSameReadingAcrossSource(incoming, candidate)) return true; + } + for (const candidate of this.inBatchCandidates) { + if (isSameReadingAcrossSource(incoming, candidate)) return true; + } + return false; + } + + private isExactStoredDuplicate(p: PreparedEntry): boolean { + return ( + sampleIsStoredDuplicate( + this.stored.existingByExternalKey.get( + externalKeyOf(p.row.type, p.row.source, p.row.externalId), + ), + p.row, + ) && + this.stored.onceInThisBatch(p.row) && + classifyRangeRejection( + p.row.type as MeasurementType, + p.row.value as number, + ) === null + ); + } + + private isUnchangedStatsRow(p: PreparedEntry): boolean { + return ( + this.stored.onceInThisBatch(p.row) && + classifyRangeRejection( + p.row.type as MeasurementType, + p.row.value as number, + ) === null && + statsWriteUnchanged( + this.stored.storedStats.get( + externalKeyOf(p.row.type, p.row.source, p.row.externalId), + ), + desiredFor(this.userId, p), + ) + ); + } + + private pushMergeCandidate(p: PreparedEntry): void { + if (isMergeableSource(p.row.source as string)) { + this.inBatchCandidates.push({ + type: p.row.type as MeasurementType, + source: p.row.source as string, + value: p.row.value as number, + measuredAt: p.row.measuredAt as Date, + }); + } + } + + /** + * Record one row's verdict. `candidate` is false for a row whose merge + * candidate was already pushed when it was buffered (see below). + */ + private applyVerdict( + p: PreparedEntry, + verdict: Verdict, + candidate: boolean, + ): void { + switch (verdict.status) { + case "inserted": + this.results[p.index] = { index: p.index, status: "inserted" }; + this.insertedCount++; + this.insertedPrepared.push(p); + this.writtenIdentities.push({ + type: p.row.type as MeasurementType, + measuredAt: p.row.measuredAt as Date, + }); + break; + case "updated": + case "resurrected": + this.results[p.index] = { index: p.index, status: "updated" }; + this.updatedCount++; + this.writtenIdentities.push( + { + type: p.row.type as MeasurementType, + measuredAt: p.row.measuredAt as Date, + }, + ...(verdict.dirtyIdentities ?? []), + ); + break; + case "duplicate": + this.results[p.index] = { index: p.index, status: "duplicate" }; + this.duplicateCount++; + break; + case "failed": + this.results[p.index] = { + index: p.index, + status: "failed", + reason: verdict.error?.code ?? "write_failed", + }; + this.failedCount++; + return; + case "rejected_range": + // The reconciler carries the same plausibility gate the first pass + // already applied, 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. + this.results[p.index] = outOfRangeResult( + p.index, + p.row.type as string, + p.row.value as number, + ); + return; + } + if (candidate) this.pushMergeCandidate(p); + } + + private reconcileRow(p: PreparedEntry) { + return prisma.$transaction((tx) => + reconcileExternalMeasurement(tx, desiredFor(this.userId, p), { + exactExternalMatch: isStatsExternalId(p.row.externalId as string) + ? "update" + : "duplicate", + }), + ); + } + + // Set-based inserts. A fresh sample used to cost one interactive + // transaction of about ten round trips (savepoint, three advisory locks, + // two candidate reads, the insert), so a 500-entry batch of new samples + // was 500 of those in a row: the 2–15 s outliers. + // + // The reconciler inserts exactly when no row, live or tombstoned, holds + // the entry's external identity `(type, source, externalId)` or its + // natural identity `(type, measuredAt, source, sleepStage)`. Both are + // full unique indexes (the natural one NULLS NOT DISTINCT), so + // `INSERT … ON CONFLICT DO NOTHING` inserts exactly those rows and skips + // every other; a skipped row then goes through the reconciler as before + // and gets its own verdict (update, resurrect, duplicate). The insert + // runs under the same per-account advisory lock the reconciler takes, so + // the two serialise like two reconcilers would. + // + // Only a row whose external and natural key each occur once in the batch + // is buffered, and the buffer is flushed before any other row reaches the + // reconciler. Such a row shares no identity with another entry, so no + // other entry's reconcile can create, move or retire a row at its + // identity, and inserting it out of turn changes nothing. Rows the insert + // skips are reconciled in their original order. A buffered row's merge + // candidate is pushed when it is buffered, as the sequential loop pushed + // it after a successful write; the only case that differs is a write that + // fails outright, where the candidate stays. + private async flushPendingInserts(): Promise { + if (this.pendingInserts.length === 0) return; + const batch = this.pendingInserts.splice(0, this.pendingInserts.length); + const landed = new Set(); + try { + for (let i = 0; i < batch.length; i += SET_INSERT_CHUNK) { + const slice = batch.slice(i, i + SET_INSERT_CHUNK); + const rows = await prisma.$transaction(async (tx) => { + await lockMeasurementUser(tx, this.userId); + return tx.measurement.createManyAndReturn({ + data: slice.map((p) => ({ + ...desiredFor(this.userId, p), + sleepStage: (p.row.sleepStage ?? + null) as Prisma.MeasurementUncheckedCreateInput["sleepStage"], + deletedAt: null, + })), + skipDuplicates: true, + select: { type: true, source: true, externalId: true }, + }); + }); + const keys = new Set( + rows.map((r) => `${r.type}::${r.source}::${r.externalId}`), + ); + for (const p of slice) { + if (keys.has(`${p.row.type}::${p.row.source}::${p.row.externalId}`)) + landed.add(p); + } + } + } catch (err) { + // Whatever did not land goes through the reconciler below, which + // turns a real database fault into a per-entry `failed` verdict. + annotate({ + action: { name: "measurement.batch.set_insert_fallback" }, + meta: { + error_class: err instanceof Error ? err.name : "unknown", + landed: landed.size, + pending: batch.length, + }, + }); + } + this.setInsertedCount += landed.size; + for (const p of batch) { + if (landed.has(p)) this.applyVerdict(p, { status: "inserted" }, false); + else this.applyVerdict(p, await this.reconcileRow(p), false); + } + } +}