diff --git a/src/__tests__/dense-measurement-read-guard.test.ts b/src/__tests__/dense-measurement-read-guard.test.ts index 086a54cb3..1134cdf0d 100644 --- a/src/__tests__/dense-measurement-read-guard.test.ts +++ b/src/__tests__/dense-measurement-read-guard.test.ts @@ -67,6 +67,8 @@ const SCOPED_DIRS = [ // Request-handler logic moved out of its route into a service keeps the // route's scope. "lib/measurements/batch-ingest/", + "lib/measurements/create/", + "lib/measurements/list/measurement-list.ts", ]; /** Types a device writes at sampling rate rather than a few times a day. */ @@ -108,7 +110,7 @@ const ALLOWED: Record = { "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.", "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)": + "lib/measurements/create/create-sleep-stage.ts::(shorthand)": "Cross-source merge probe for one posted sleep segment: SLEEP_DURATION in a ±tolerance window around its end.", "lib/targets/build-response.ts::{ in: recentTypes": "Thirty days of weight, blood pressure, resting heart rate, body fat and steps (drained to one row per day). PULSE is filtered out of `recentTypes` and folds in SQL.", diff --git a/src/__tests__/measurement-write-plausibility-guard.test.ts b/src/__tests__/measurement-write-plausibility-guard.test.ts index a26ebc4ca..12c9de285 100644 --- a/src/__tests__/measurement-write-plausibility-guard.test.ts +++ b/src/__tests__/measurement-write-plausibility-guard.test.ts @@ -68,7 +68,8 @@ const WRITERS: Record = { "lib/measurements/reconcile-external-measurement.ts": { kind: "gate" }, // ── Person-driven write paths, validated ────────────────────────────── - "app/api/measurements/route.ts": { + // The hand-entry writes behind `POST /api/measurements`. + "lib/measurements/create/create-readings.ts": { kind: "upstream", validatedIn: "lib/validations/measurement.ts", }, diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 24e03e639..a9befc1ff 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -54,7 +54,7 @@ const WRITE_SITES: Record = { "lib/measurements/batch-ingest/write-measurements.ts": TAIL_VIA( "lib/measurements/batch-ingest/tail.ts", ), - "app/api/measurements/route.ts": TAIL, + "lib/measurements/create/create-readings.ts": TAIL, "app/api/mental-health/assessments/route.ts": TAIL, "lib/insights/score-row.ts": TAIL, "lib/mcp/writes.ts": TAIL, diff --git a/src/app/api/measurements/route.ts b/src/app/api/measurements/route.ts index 655666a85..1a142980f 100644 --- a/src/app/api/measurements/route.ts +++ b/src/app/api/measurements/route.ts @@ -1,13 +1,10 @@ -import { prisma } from "@/lib/db"; import { apiHandler, isScopedCredential, requireRecordAuth, } 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 { fireAndForget } from "@/lib/logging/fire-and-forget"; import { auditLog } from "@/lib/auth/audit"; import { apiSuccess, @@ -21,52 +18,24 @@ import { createMeasurementSchema, createBatchMeasurementSchema, listMeasurementsSchema, - getUnitForType, } from "@/lib/validations/measurement"; -import { - BUCKET_CAP, - type AggregateGrain, -} from "@/lib/measurements/range-aggregation"; -import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; -import { localDayWindow } from "@/lib/tz/local-day"; import { withIdempotency } from "@/lib/idempotency"; -import { isP2002 } from "@/lib/prisma-errors"; -import { encryptNote, shapeMeasurementNotes } from "@/lib/crypto/note-cipher"; -import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; -import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; -import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; -import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; -import { runSafetyFloorCheck } from "@/lib/illness/safety-floor-check"; -import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; -import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { shapeMeasurementNotes } from "@/lib/crypto/note-cipher"; +import { measurementListResponse } from "@/lib/measurements/list/measurement-list"; import { - readDailySeries, - readLiveBuckets, -} from "@/lib/measurements/daily-series-read"; -import { dayGroupedListResponse } from "@/lib/measurements/list/day-grouped"; -import { sleepListResponse } from "@/lib/measurements/list/sleep-list"; -import { resolveStoredTimezone, resolveUserTimezone } from "@/lib/tz/resolver"; + createReading, + createReadings, + type ReadingWriter, +} from "@/lib/measurements/create/create-readings"; +import { createNamedSleepStage } from "@/lib/measurements/create/create-sleep-stage"; import { isNamedSleepStageBody, sleepStageEntrySchema, - toSleepStageSegment, } from "@/lib/validations/sleep-stage-entry"; -import { - MeasurementReconciliationError, - reconcileExternalMeasurement, -} from "@/lib/measurements/reconcile-external-measurement"; -import { - MEASURED_AT_TOLERANCE_MS, - isMergeableSource, - isSameReadingAcrossSource, - oppositeMergeSource, -} from "@/lib/measurements/cross-source-merge"; import { NextRequest } from "next/server"; -import type { - MeasurementType, - MeasurementSource, - GlucoseContext, -} from "@/generated/prisma/client"; + +const DUPLICATE_MESSAGE = "A measurement with this data already exists"; +const DUPLICATE_META = { errorCode: "measurement.duplicate_timestamp" }; export const GET = apiHandler(async (request: NextRequest) => { const { user } = await requireRecordAuth("read", "measurements"); @@ -102,257 +71,7 @@ export const GET = apiHandler(async (request: NextRequest) => { return returnAllZodIssues(parsed.error, 422); } - const { - type, - from, - to, - limit, - offset, - sortBy, - sortDir, - aggregate, - source, - sourceEq, - valueMin, - valueMax, - groupBy, - dayKey, - } = parsed.data; - - const where = { - userId: user.id, - // v1.4.40 — soft-deleted readings are invisible to every - // list / count / aggregate read. The dedicated `sync/state` route - // keeps surfacing them via its own filter so iOS can reconcile - // tombstones; every other consumer (list view, analytics fan-out, - // dashboard read paths) must never include them. - deletedAt: null, - ...(type && { type: type as MeasurementType }), - // v1.15.13 — management-list source filter. Distinct from the - // `source=rollup` opt-in above (which selects the rollup read tier); - // `sourceEq` narrows the plain list to one ingest source. - ...(sourceEq && { source: sourceEq }), - // v1.18.5 — value-range filter (backlog G). Open-ended on either side; - // the validator has already rejected an inverted `valueMin > valueMax`. - ...(valueMin != null || valueMax != null - ? { - value: { - ...(valueMin != null && { gte: valueMin }), - ...(valueMax != null && { lte: valueMax }), - }, - } - : {}), - ...(from || to - ? { - measuredAt: { - ...(from && { gte: from }), - ...(to && { lte: to }), - }, - } - : {}), - }; - - // v1.11.5 — sleep is read row-per-STAGE (≈5 rows/night × source-count), - // not a spot sample and not a cumulative day-sum. A plain list query for - // SLEEP_DURATION would leak every stage segment as its own "value" — the - // "many small numbers that stand for themselves" complaint. Route it - // through the canonical per-night aggregation (the same helper the chart - // series + dashboard tile use) so a sleep list returns one row per night - // carrying the night's TIME ASLEEP, with a `dayKey` drill-down to that - // night's stage segments. Storage is untouched; only the display - // collapses. Runs ahead of the cumulative branches because sleep is in - // neither CUMULATIVE_HK_TYPES nor the spot-sample set. - if (type === "SLEEP_DURATION") { - return sleepListResponse(user, { - from, - to, - limit, - offset, - sortDir, - dayKey, - }); - } - - // v1.4.37 — list-view drill-down to per-sample rows for a single - // calendar day in the user's IANA timezone. Resolves the day's UTC - // bounds against `User.timezone`, then runs a plain `findMany` so the - // expandable list row shows every per-sample chunk the daily collapse - // hid. Gated to cumulative HK types — the spot-sample types render - // one-row-per-event already and have nothing to drill into. - if (dayKey && type && CUMULATIVE_HK_TYPES.has(type as MeasurementType)) { - // The same zone the day-grouped list cuts its days in, so a row's - // drill-down opens exactly the day the row summed. - const tz = await resolveStoredTimezone(user.timezone); - // Resolve `[dayStart, dayEnd)` for the requested calendar day in - // the user's local zone. v1.4.37 — `localDayWindow` honours - // DST: the previous shape (`canonicalDailyTimestamp ± 12 h`) - // silently leaked or hid one hour of samples on spring-forward - // (23-h day) and fall-back (25-h day) by always spanning exactly - // 24 h. Works for every IANA offset including sub-half-hour - // zones (Asia/Kathmandu UTC+5:45, Pacific/Chatham UTC+12:45). - const { dayStart, dayEnd } = localDayWindow(dayKey, tz); - const samples = await prisma.measurement.findMany({ - where: { - userId: user.id, - type: type as MeasurementType, - measuredAt: { gte: dayStart, lt: dayEnd }, - deletedAt: null, - }, - // Same tiebreaker as the paged list. This one slices in memory rather - // than with `skip`, so there is no page drift to prevent — but two - // identical requests would still be free to render a tied block in - // different orders, and a drill-down that reshuffles on refresh reads - // as data changing. - orderBy: [{ measuredAt: sortDir }, { id: sortDir }], - // v1.4.38 — the 1000-row cap now lives on the validator (refine - // on `(limit, dayKey)` returning 422 when a caller asks for - // more). The route reads `limit` straight through because the - // validator has already clamped it. - take: limit, - }); - annotate({ - action: { name: "measurement.list" }, - meta: { total: samples.length, type, dayKey, mode: "drill-down" }, - }); - return apiSuccess({ - measurements: samples, - meta: { total: samples.length, limit, offset: 0, dayKey }, - }); - } - - // v1.4.37 — collapsed "one row per day" mode for cumulative - // types. The list page renders the day's SUM as a single synthesised - // row with a `sampleCount` chevron that drills back into the - // per-sample rows via the `dayKey=…` branch above. `limit` / `offset` - // page over DAYS, not raw rows (#1245). - if ( - groupBy === "day" && - type && - CUMULATIVE_HK_TYPES.has(type as MeasurementType) - ) { - return dayGroupedListResponse(user, { - type: type as MeasurementType, - limit, - offset, - sortDir, - sourceEq, - from, - to, - }); - } - - // v1.4.28 — server-side aggregation. Gated on an explicit - // `aggregate` param so the iOS contract (raw `MeasurementWireDTO` - // shape on `GET /api/measurements`) is byte-stable for any caller - // that omits the new query parameter. The chart-data client opts - // in; iOS does not. - // - // `take` no longer applies BEFORE bucketising. The - // aggregation runs in Postgres via `date_trunc`, the bucket cap is - // applied AFTER, so a 1-year `aggregate=daily` window walks every - // row in the window and returns up to 365 buckets per type instead - // of truncating to the first N raw rows. - // - // "All time" range semantics. When the chart's "All" tab - // is active, the client passes `from` = the user's earliest - // measurement (or a sentinel like 1970-01-01) plus `to` = now plus - // `aggregate=monthly` (or `weekly` when full history < 2 years). - // The route's response is bounded by the `BUCKET_CAP` ceiling per - // grain (monthly: 24, weekly: 105, daily: 365) so a multi-decade - // account never paints an unbounded series. - // `source=rollup` + `aggregate=daily` — the chart's series read. Windows up - // to the DAY cap fold live into the user's local days (#1026); wider - // windows step up to the WEEK / MONTH / YEAR rollup tier. `type` is - // mandatory so the response mirrors the per-type ask the chart makes. - if (source === "rollup" && aggregate === "daily" && type && from && to) { - // Shared with the batched series endpoint, so both read the same rows. - const cap = Math.min(limit, BUCKET_CAP.daily); - const priorityJson = await loadUserSourcePriority(user.id); - const measurements = await readDailySeries({ - userId: user.id, - type: type as MeasurementType, - from, - to, - limit, - priorityJson, - }); - if (measurements.length > 0) { - return apiSuccess({ - measurements, - meta: { - total: measurements.length, - limit: cap, - offset: 0, - aggregate: "daily", - }, - }); - } - // An empty result means the window genuinely holds no rows. Fall - // through to the shared live aggregate branch below to keep the - // response shape identical for the truly-empty case. - } - - if (aggregate && aggregate !== "raw" && from && to) { - const grain: Exclude = aggregate; - const cap = Math.min(limit, BUCKET_CAP[grain]); - // Buckets are cut at the user's own calendar boundaries (#1026), the - // same day the raw list and the chart's short range use. Cumulative - // types SUM per bucket, every other type averages; overlapping sources - // collapse to the ladder-canonical one first. - const priorityJson = await loadUserSourcePriority(user.id); - const measurements = await readLiveBuckets({ - userId: user.id, - type: (type as MeasurementType | undefined) ?? null, - from, - to, - cap, - priorityJson, - grain, - timeZone: await resolveUserTimezone(user.id), - }); - annotate({ - action: { name: "measurement.list" }, - meta: { total: measurements.length, type, aggregate: grain }, - }); - return apiSuccess({ - measurements, - meta: { - total: measurements.length, - limit: cap, - offset: 0, - aggregate: grain, - }, - }); - } - - const [measurements, total] = await Promise.all([ - prisma.measurement.findMany({ - where, - // `{ id: sortDir }` is the tiebreaker, not decoration. `measuredAt` - // is not unique here and ties are the normal case. Only the exact - // tuple `(userId, type, measuredAt, source, sleepStage)` is unique, - // so a tie survives across `type` (the `stats::YYYY-MM-DD` - // daily-rollup convention puts every daily total at one instant), - // across `source` (two writers landing on the same instant) and - // across `sleepStage`. Without a unique secondary key Postgres is - // free to order the tied block differently between the query for page - // N and the query for page N+1, so a client paging its own history - // silently gets one row twice and never gets another. - orderBy: [{ [sortBy]: sortDir }, { id: sortDir }], - take: limit, - skip: offset, - }), - prisma.measurement.count({ where }), - ]); - - annotate({ action: { name: "measurement.list" }, meta: { total, type } }); - - return apiSuccess({ - // v1.23 — decrypt `notesEncrypted` onto `notes` and strip the ciphertext - // before it leaves the server (iOS MeasurementWireDTO contract). - measurements: measurements.map(shapeMeasurementNotes), - meta: { total, limit, offset }, - }); + return measurementListResponse(user, parsed.data); }); /** @@ -432,11 +151,17 @@ async function postMeasurement(request: NextRequest) { ); } + const writer: ReadingWriter = { + user: { id: user.id, timezone: user.timezone }, + scoped, + ipAddress: getClientIp(request), + }; + // One sleep-stage segment, the stage named rather than coded. A bridge // (Tasker, Home Assistant) posts here with its narrow token; the row and // its dedupe are the batch route's. See `sleep-stage-entry.ts`. if (isNamedSleepStageBody(body)) { - return postNamedSleepStage(request, user.id, body, scoped); + return postNamedSleepStage(writer, body); } // Batch mode (array of measurements, e.g. combined BP + Pulse) @@ -473,136 +198,11 @@ async function postMeasurement(request: NextRequest) { }); } - // The whole batch is one transaction, so a duplicate anywhere in it rolls - // the rest back — and the caller has to be told which condition it hit. - // Untouched, that duplicate escaped as an unhandled error and the batch - // answered 500, which is indistinguishable from a server fault and is what - // every client retries against. - let results; - try { - results = await prisma.$transaction( - parsed.data.measurements.map((m) => - prisma.measurement.create({ - data: { - userId: user.id, - type: m.type as MeasurementType, - value: m.value, - unit: getUnitForType(m.type), - // v1.38.x — the array arm too, not just the single create - // below. This is the combined BP + Pulse path a bridge uses for - // a two-value reading; missing it here would leave exactly those - // rows attributed MANUAL and invisible to the badge and the - // source filter. `scoped` already implies `m.source` is - // undefined — the refusal above returned otherwise — so this is - // the resolution, not an override. - source: (scoped - ? EXTERNAL_SOURCE - : (m.source ?? "MANUAL")) as MeasurementSource, - measuredAt: m.measuredAt, - // v1.23 — encrypt the note at rest; the legacy plaintext column is - // written null for new rows. - notes: null, - notesEncrypted: encryptNote(m.notes ?? null), - glucoseContext: - (m.glucoseContext as GlucoseContext | undefined) ?? null, - // v1.4.25 (code-review M4): mirror the - // single-entry path so the multi-entry batch persists - // `deviceType` instead of silently dropping it. - deviceType: m.deviceType ?? null, - }, - }), - ), - ); - } catch (err) { - if (isP2002(err)) { - return apiError("A measurement with this data already exists", 409, { - errorCode: "measurement.duplicate_timestamp", - }); - } - throw err; + const created = await createReadings(writer, parsed.data); + if ("duplicate" in created) { + return apiError(DUPLICATE_MESSAGE, 409, DUPLICATE_META); } - - const shapedResults = results.map(shapeMeasurementNotes); - - await auditLog("measurement.create.batch", { - userId: user.id, - ipAddress: getClientIp(request), - details: { - count: results.length, - types: parsed.data.measurements.map((m) => m.type), - }, - }); - - annotate({ - action: { name: "measurement.create.batch" }, - meta: { - count: results.length, - types: parsed.data.measurements.map((m) => m.type), - }, - }); - - // v1.4.34 — flush every cache that reflects this user's - // measurement set so the next read paints the new rows. Interactive - // multi-entry create (combined BP + pulse form) — hard-evict so the - // SWR readers don't serve the pre-write body back to the user. - invalidateUserMeasurements(user.id, { evict: true }); - - // v1.18.1 — eventful Vorsorge satisfaction. Resolve the user's - // reminders against the just-landed readings now. Fire-and-forget. - fireAndForget(enqueueReminderSatisfy(user.id), { - action: "reminder.satisfy.enqueue", - }); - - // v1.31.0 — feed the shared arrival seams only the rows returned by the - // create-only transaction. Both callbacks are best-effort: arrival or - // morning-refresh infrastructure can never turn a committed write into an - // error response. - void emitInsertedMeasurementArrivals(user.id, results, "manual").catch( - () => {}, - ); - const sleepMeasuredAts = results - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt); - if (sleepMeasuredAts.length > 0) { - void maybeEnqueueMorningRefresh(user.id, sleepMeasuredAts).catch( - () => {}, - ); - } - - // v1.18.6 — absolute clinical safety-floor check on the just-written - // readings (confirm-gated, module-gated, never diagnoses). The combined - // BP form lands both arms here, so the BP floor sees a whole reading. - // The transient per-entry `symptomsPresent` flag (ORed across the batch) - // lifts a confirmed breach to the symptom-coupled emergency copy. - // Fire-and-forget — a notification failure never fails the write. - fireAndForget( - runSafetyFloorCheck({ - userId: user.id, - written: results.map((r) => ({ - type: r.type, - value: r.value, - measuredAt: r.measuredAt, - glucoseContext: r.glucoseContext, - })), - symptomsPresent: parsed.data.measurements.some( - (m) => m.symptomsPresent === true, - ), - timezone: user.timezone ?? undefined, - }), - { action: "measurement.safety_floor.check" }, - ); - - // v1.37.19 — the shared post-mutation tail: rollup recompute - // (collapsed per (type, day)) + status-insight re-warm, both - // best-effort. One helper so a future write surface cannot remember - // one leg and forget the other. - await afterMeasurementMutation( - user.id, - results.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), - { source: "manual" }, - ); - - return apiSuccess(shapedResults, 201); + return apiSuccess(created.rows.map(shapeMeasurementNotes), 201); } // Single mode (existing behavior) @@ -633,155 +233,20 @@ async function postMeasurement(request: NextRequest) { }); } - const { - type, - value, - measuredAt, - notes, - source, - glucoseContext, - deviceType, - symptomsPresent, - } = parsed.data; - - // Handle unique constraint violation - let measurement; - try { - measurement = await prisma.measurement.create({ - data: { - userId: user.id, - type: type as MeasurementType, - value, - unit: getUnitForType(type), - // v1.38.x — server-resolved under a scoped credential. `source` is - // necessarily the schema default here (the refusal above returned on - // any body that named one), so nothing a caller asserted is being - // discarded. - source: (scoped - ? EXTERNAL_SOURCE - : (source ?? "MANUAL")) as MeasurementSource, - measuredAt, - // v1.23 — encrypt the note at rest; legacy plaintext column nulled. - notes: null, - notesEncrypted: encryptNote(notes ?? null), - glucoseContext: (glucoseContext as GlucoseContext | undefined) ?? null, - // v1.4.25 (code-review M4): the deviceType column - // already accepts client metadata from the batch route. Mirror - // the behaviour here so single-entry POST persists the tag - // instead of silently dropping it. - deviceType: deviceType ?? null, - }, - }); - } catch (err) { - // `isP2002` rather than `instanceof Prisma.PrismaClientKnownRequestError`. - // The helper's own docblock says why it exists: the structural check - // survives client-bundling quirks, and the `instanceof` form does not — - // when it misses, a duplicate escapes as an unhandled error and the caller - // gets a 500 for a condition the route has always meant to answer with a - // 409. A retry loop that meets a 500 keeps retrying. - if (isP2002(err)) { - return apiError("A measurement with this data already exists", 409, { - errorCode: "measurement.duplicate_timestamp", - }); - } - throw err; + const created = await createReading(writer, parsed.data); + if ("duplicate" in created) { + return apiError(DUPLICATE_MESSAGE, 409, DUPLICATE_META); } - - await auditLog("measurement.create", { - userId: user.id, - ipAddress: getClientIp(request), - details: { type, measurementId: measurement.id }, - }); - - annotate({ - action: { - name: "measurement.create", - entity_type: "measurement", - entity_id: measurement.id, - }, - }); - - // v1.4.34 — flush every cache that reflects this user's - // measurement set so the next read paints the new row. Interactive - // single-entry create — hard-evict so the SWR readers don't serve the - // pre-write body back to the user. - invalidateUserMeasurements(user.id, { evict: true }); - - // v1.18.1 — eventful Vorsorge satisfaction. Resolve the user's reminders - // against the just-landed reading now. Fire-and-forget. - fireAndForget(enqueueReminderSatisfy(user.id), { - action: "reminder.satisfy.enqueue", - }); - - // v1.31.0 — `measurement` is the exact successful create result. A P2002 - // duplicate returns above and never reaches either best-effort callback. - void emitInsertedMeasurementArrivals(user.id, [measurement], "manual").catch( - () => {}, - ); - if (measurement.type === "SLEEP_DURATION") { - void maybeEnqueueMorningRefresh(user.id, [measurement.measuredAt]).catch( - () => {}, - ); - } - - // v1.18.6 — absolute clinical safety-floor check (confirm-gated, - // module-gated, never diagnoses). A single-entry POST carries only one arm - // of a BP reading, so the BP floor only fires through the combined-form / - // batch path; a lone glucose reading is evaluated here. The transient - // `symptomsPresent` flag lifts a confirmed breach to the emergency copy. - // Fire-and-forget — a notification failure never fails the write. - fireAndForget( - runSafetyFloorCheck({ - userId: user.id, - written: [ - { - type: measurement.type, - value: measurement.value, - measuredAt: measurement.measuredAt, - glucoseContext: measurement.glucoseContext, - }, - ], - symptomsPresent, - timezone: user.timezone ?? undefined, - }), - { action: "measurement.safety_floor.check" }, - ); - - // v1.5.0 — refresh the persistent rollup row for the affected - // (type, day) tuple. Runs inline so the next read of the - // comprehensive / analytics surface is correct without waiting on - // pg-boss; WEEK / MONTH / YEAR recomputes are enqueued under the - // hood. Best-effort — a populator hiccup never fails the user's - // write. - await afterMeasurementMutation( - user.id, - [{ type: measurement.type, measuredAt: measurement.measuredAt }], - { source: "manual" }, - ); - - return apiSuccess(shapeMeasurementNotes(measurement), 201); + return apiSuccess(shapeMeasurementNotes(created.row), 201); } /** - * `POST /api/measurements` with a named `sleepStage`: one segment, stored as - * the batch route stores a HealthKit sleep sample. - * - * The same three steps as the batch route, in the same order: a same-reading - * row of the opposite source (MANUAL against APPLE_HEALTH) makes this one a - * duplicate; otherwise `reconcileExternalMeasurement` settles both unique - * identities at once, reporting an exact external-id match as a duplicate - * and adopting a row that only shares the natural identity. So a bridge that - * re-posts the same segment gets one row, whichever path it used first. - * - * `201` with the stored row when it was inserted, `200` when it was already - * there (`duplicate`) or was corrected in place (`updated`). The row carries - * a `status` saying which. + * The named sleep-stage arm: validated here, stored by + * `createNamedSleepStage` (see its header for the dedupe and the answers). */ async function postNamedSleepStage( - request: NextRequest, - userId: string, + writer: ReadingWriter, body: unknown, - scoped: boolean, ): Promise { const parsed = sleepStageEntrySchema.safeParse(body); if (!parsed.success) { @@ -793,146 +258,12 @@ async function postNamedSleepStage( errorCode: "measurement.create.invalid", }); } - - const segment = toSleepStageSegment(parsed.data); - // Under a scoped credential the source is resolved from the credential, - // as on the other arms of this route: the refusal above already returned - // on any body that named one. - const source = ( - scoped ? EXTERNAL_SOURCE : parsed.data.source - ) as MeasurementSource; - const type: MeasurementType = "SLEEP_DURATION"; - - // Cross-source same reading, exactly as the batch route decides it. - if (isMergeableSource(source)) { - const candidates = await prisma.measurement.findMany({ - where: { - userId, - deletedAt: null, - type, - source: oppositeMergeSource(source), - measuredAt: { - gte: new Date(segment.end.getTime() - MEASURED_AT_TOLERANCE_MS), - lte: new Date(segment.end.getTime() + MEASURED_AT_TOLERANCE_MS), - }, - }, - select: { - id: true, - type: true, - source: true, - value: true, - measuredAt: true, - }, - }); - const twin = candidates.find((candidate) => - isSameReadingAcrossSource( - { type, source, value: segment.minutes, measuredAt: segment.end }, - candidate, - ), - ); - if (twin) { - return namedSleepStageResponse(twin.id, "duplicate", { - stage: segment.stage, - source, - crossSource: true, - }); - } - } - - const verdict = await prisma.$transaction((tx) => - reconcileExternalMeasurement( - tx, - { - userId, - type, - value: segment.minutes, - unit: getUnitForType(type), - source, - measuredAt: segment.end, - externalId: segment.externalId, - externalSourceVersion: null, - sleepStage: segment.stage, - deviceType: parsed.data.deviceType ?? null, - }, - { exactExternalMatch: "duplicate" }, - ), - ); - - if (verdict.status === "failed") { - throw new MeasurementReconciliationError(verdict); - } - if (verdict.status === "rejected_range") { - // The schema applies the same band, so this is the two gates disagreeing. - return apiError("Value out of plausible range", 422, { - errorCode: "measurement.create.invalid", - }); - } - - const status = - verdict.status === "inserted" - ? "inserted" - : verdict.status === "duplicate" - ? "duplicate" - : "updated"; - - if (status !== "duplicate") { - await auditLog("measurement.create", { - userId, - ipAddress: getClientIp(request), - details: { - type, - measurementId: verdict.row.id, - sleepStage: segment.stage, - status, - }, - }); - invalidateUserMeasurements(userId, { evict: true }); - fireAndForget(enqueueReminderSatisfy(userId), { - action: "reminder.satisfy.enqueue", - }); - if (status === "inserted") { - void emitInsertedMeasurementArrivals( - userId, - [verdict.row], - "manual", - ).catch(() => {}); - void maybeEnqueueMorningRefresh(userId, [segment.end]).catch(() => {}); - } - await afterMeasurementMutation( - userId, - [{ type, measuredAt: segment.end }, ...(verdict.dirtyIdentities ?? [])], - { source: "manual" }, - ); - } - - return namedSleepStageResponse(verdict.row.id, status, { - stage: segment.stage, - source, - crossSource: false, - }); -} - -async function namedSleepStageResponse( - id: string, - status: "inserted" | "updated" | "duplicate", - meta: { stage: string; source: string; crossSource: boolean }, -): Promise { - annotate({ - action: { - name: "measurement.create.sleep-stage", - entity_type: "measurement", - entity_id: id, + return createNamedSleepStage( + { + userId: writer.user.id, + scoped: writer.scoped, + ipAddress: writer.ipAddress, }, - meta: { - status, - stage: meta.stage, - source: meta.source, - cross_source_merge: meta.crossSource, - }, - }); - const row = await prisma.measurement.findUniqueOrThrow({ where: { id } }); - return apiSuccess( - { ...shapeMeasurementNotes(row), status }, - status === "inserted" ? 201 : 200, + parsed.data, ); } diff --git a/src/lib/measurements/create/create-readings.ts b/src/lib/measurements/create/create-readings.ts new file mode 100644 index 000000000..0ab7ea82d --- /dev/null +++ b/src/lib/measurements/create/create-readings.ts @@ -0,0 +1,259 @@ +/** + * The hand-entry writes behind `POST /api/measurements`: one reading, or a + * small array of them written together (the combined blood-pressure and + * pulse form, a dual-value glucose entry), with the work a fresh reading + * starts. The route owns authentication, the body, the source refusal and + * the validation answers; these return what was written, or that the write + * met a stored duplicate. + */ +import type { z } from "zod/v4"; + +import { prisma } from "@/lib/db"; +import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source"; +import { annotate } from "@/lib/logging/context"; +import { fireAndForget } from "@/lib/logging/fire-and-forget"; +import { auditLog } from "@/lib/auth/audit"; +import { + getUnitForType, + type createBatchMeasurementSchema, + type createMeasurementSchema, +} from "@/lib/validations/measurement"; +import { isP2002 } from "@/lib/prisma-errors"; +import { encryptNote } from "@/lib/crypto/note-cipher"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; +import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; +import { runSafetyFloorCheck } from "@/lib/illness/safety-floor-check"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import type { + GlucoseContext, + Measurement, + MeasurementSource, + MeasurementType, +} from "@/generated/prisma/client"; + +/** Who the reading is for and how the request authenticated. */ +export interface ReadingWriter { + /** The record the reading belongs to (the owner under a delegation). */ + user: { id: string; timezone: string | null }; + /** A narrow bridge credential: its rows are `EXTERNAL`. */ + scoped: boolean; + ipAddress: string | null; +} + +/** A unique index already holds this reading. */ +export const DUPLICATE_READING = { duplicate: true } as const; + +/** + * The array arm: every reading in one transaction, so a duplicate anywhere + * in it rolls the rest back. + */ +export async function createReadings( + writer: ReadingWriter, + input: z.output, +): Promise<{ rows: Measurement[] } | typeof DUPLICATE_READING> { + const { user, scoped } = writer; + // The caller has to be told which condition the batch hit. Untouched, a + // duplicate escaped as an unhandled error and the batch answered 500, + // which is indistinguishable from a server fault and is what every client + // retries against. + let results; + try { + results = await prisma.$transaction( + input.measurements.map((m) => + prisma.measurement.create({ + data: { + userId: user.id, + type: m.type as MeasurementType, + value: m.value, + unit: getUnitForType(m.type), + // The array arm too, not just the single create: this is the + // combined BP + Pulse path a bridge uses for a two-value reading; + // missing it here would leave exactly those rows attributed + // MANUAL and invisible to the badge and the source filter. + // `scoped` already implies `m.source` is undefined — the route + // refused otherwise — so this is the resolution, not an override. + source: (scoped + ? EXTERNAL_SOURCE + : (m.source ?? "MANUAL")) as MeasurementSource, + measuredAt: m.measuredAt, + // The note is sealed at rest; the legacy plaintext column is + // written null for new rows. + notes: null, + notesEncrypted: encryptNote(m.notes ?? null), + glucoseContext: + (m.glucoseContext as GlucoseContext | undefined) ?? null, + // Mirror the single-entry path so the multi-entry batch persists + // `deviceType` instead of silently dropping it. + deviceType: m.deviceType ?? null, + }, + }), + ), + ); + } catch (err) { + if (isP2002(err)) return DUPLICATE_READING; + throw err; + } + + await auditLog("measurement.create.batch", { + userId: user.id, + ipAddress: writer.ipAddress, + details: { + count: results.length, + types: input.measurements.map((m) => m.type), + }, + }); + + annotate({ + action: { name: "measurement.create.batch" }, + meta: { + count: results.length, + types: input.measurements.map((m) => m.type), + }, + }); + + await afterReadingsCreated(user, results, { + symptomsPresent: input.measurements.some((m) => m.symptomsPresent === true), + }); + return { rows: results }; +} + +/** The single-reading arm. */ +export async function createReading( + writer: ReadingWriter, + input: z.output, +): Promise<{ row: Measurement } | typeof DUPLICATE_READING> { + const { user, scoped } = writer; + const { + type, + value, + measuredAt, + notes, + source, + glucoseContext, + deviceType, + symptomsPresent, + } = input; + + let measurement; + try { + measurement = await prisma.measurement.create({ + data: { + userId: user.id, + type: type as MeasurementType, + value, + unit: getUnitForType(type), + // Server-resolved under a scoped credential. `source` is necessarily + // the schema default here (the route refused any body that named + // one), so nothing a caller asserted is being discarded. + source: (scoped + ? EXTERNAL_SOURCE + : (source ?? "MANUAL")) as MeasurementSource, + measuredAt, + // The note is sealed at rest; the legacy plaintext column is nulled. + notes: null, + notesEncrypted: encryptNote(notes ?? null), + glucoseContext: (glucoseContext as GlucoseContext | undefined) ?? null, + // The deviceType column already accepts client metadata from the + // batch route; the single-entry POST persists the tag too instead of + // silently dropping it. + deviceType: deviceType ?? null, + }, + }); + } catch (err) { + // `isP2002` rather than `instanceof Prisma.PrismaClientKnownRequestError`. + // The helper's own docblock says why it exists: the structural check + // survives client-bundling quirks, and the `instanceof` form does not — + // when it misses, a duplicate escapes as an unhandled error and the caller + // gets a 500 for a condition the route has always meant to answer with a + // 409. A retry loop that meets a 500 keeps retrying. + if (isP2002(err)) return DUPLICATE_READING; + throw err; + } + + await auditLog("measurement.create", { + userId: user.id, + ipAddress: writer.ipAddress, + details: { type, measurementId: measurement.id }, + }); + + annotate({ + action: { + name: "measurement.create", + entity_type: "measurement", + entity_id: measurement.id, + }, + }); + + await afterReadingsCreated(user, [measurement], { symptomsPresent }); + return { row: measurement }; +} + +/** + * The work fresh hand-entered readings start, in the order both arms have + * always run it: the cache, the reminder satisfaction, the arrival and + * morning-refresh seams, the safety-floor check, and the shared rollup and + * status tail. + */ +async function afterReadingsCreated( + user: ReadingWriter["user"], + rows: Measurement[], + options: { symptomsPresent: boolean | undefined }, +): Promise { + // Flush every cache that reflects this user's measurement set so the next + // read paints the new rows. An interactive create hard-evicts, so the SWR + // readers don't serve the pre-write body back to the user. + invalidateUserMeasurements(user.id, { evict: true }); + + // Eventful Vorsorge satisfaction. Resolve the user's reminders against the + // just-landed readings now. Fire-and-forget. + fireAndForget(enqueueReminderSatisfy(user.id), { + action: "reminder.satisfy.enqueue", + }); + + // Feed the shared arrival seams only the rows the create returned. Both + // callbacks are best-effort: arrival or morning-refresh infrastructure can + // never turn a committed write into an error response. + void emitInsertedMeasurementArrivals(user.id, rows, "manual").catch(() => {}); + const sleepMeasuredAts = rows + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt); + if (sleepMeasuredAts.length > 0) { + void maybeEnqueueMorningRefresh(user.id, sleepMeasuredAts).catch(() => {}); + } + + // The absolute clinical safety-floor check on the just-written readings + // (confirm-gated, module-gated, never diagnoses). The combined BP form + // lands both arms in one request, so the BP floor sees a whole reading; a + // single-entry POST carries one arm, so the BP floor only fires through the + // combined form, while a lone glucose reading is evaluated here. The + // transient `symptomsPresent` flag lifts a confirmed breach to the + // symptom-coupled emergency copy. Fire-and-forget — a notification failure + // never fails the write. + fireAndForget( + runSafetyFloorCheck({ + userId: user.id, + written: rows.map((r) => ({ + type: r.type, + value: r.value, + measuredAt: r.measuredAt, + glucoseContext: r.glucoseContext, + })), + symptomsPresent: options.symptomsPresent, + timezone: user.timezone ?? undefined, + }), + { action: "measurement.safety_floor.check" }, + ); + + // The shared post-mutation tail: rollup recompute (collapsed per (type, + // day)) + status-insight re-warm, both best-effort. One helper so a future + // write surface cannot remember one leg and forget the other. Runs inline + // so the next read of the comprehensive / analytics surface is correct + // without waiting on pg-boss. + await afterMeasurementMutation( + user.id, + rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), + { source: "manual" }, + ); +} diff --git a/src/lib/measurements/create/create-sleep-stage.ts b/src/lib/measurements/create/create-sleep-stage.ts new file mode 100644 index 000000000..f42b235d9 --- /dev/null +++ b/src/lib/measurements/create/create-sleep-stage.ts @@ -0,0 +1,198 @@ +/** + * `POST /api/measurements` with a named `sleepStage`: one segment, stored as + * the batch route stores a HealthKit sleep sample. + * + * The same three steps as the batch route, in the same order: a same-reading + * row of the opposite source (MANUAL against APPLE_HEALTH) makes this one a + * duplicate; otherwise `reconcileExternalMeasurement` settles both unique + * identities at once, reporting an exact external-id match as a duplicate + * and adopting a row that only shares the natural identity. So a bridge that + * re-posts the same segment gets one row, whichever path it used first. + * + * `201` with the stored row when it was inserted, `200` when it was already + * there (`duplicate`) or was corrected in place (`updated`). The row carries + * a `status` saying which. + */ +import type { z } from "zod/v4"; + +import { prisma } from "@/lib/db"; +import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source"; +import { annotate } from "@/lib/logging/context"; +import { fireAndForget } from "@/lib/logging/fire-and-forget"; +import { auditLog } from "@/lib/auth/audit"; +import { apiError, apiSuccess } from "@/lib/api-response"; +import { getUnitForType } from "@/lib/validations/measurement"; +import { shapeMeasurementNotes } from "@/lib/crypto/note-cipher"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; +import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; +import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import { + toSleepStageSegment, + type sleepStageEntrySchema, +} from "@/lib/validations/sleep-stage-entry"; +import { + MeasurementReconciliationError, + reconcileExternalMeasurement, +} from "@/lib/measurements/reconcile-external-measurement"; +import { + MEASURED_AT_TOLERANCE_MS, + isMergeableSource, + isSameReadingAcrossSource, + oppositeMergeSource, +} from "@/lib/measurements/cross-source-merge"; +import type { + MeasurementSource, + MeasurementType, +} from "@/generated/prisma/client"; + +export async function createNamedSleepStage( + args: { + userId: string; + scoped: boolean; + ipAddress: string | null; + }, + entry: z.output, +): Promise { + const { userId, scoped } = args; + const segment = toSleepStageSegment(entry); + // Under a scoped credential the source is resolved from the credential, + // as on the other arms of this route: the route already refused any body + // that named one. + const source = (scoped ? EXTERNAL_SOURCE : entry.source) as MeasurementSource; + const type: MeasurementType = "SLEEP_DURATION"; + + // Cross-source same reading, exactly as the batch route decides it. + if (isMergeableSource(source)) { + const candidates = await prisma.measurement.findMany({ + where: { + userId, + deletedAt: null, + type, + source: oppositeMergeSource(source), + measuredAt: { + gte: new Date(segment.end.getTime() - MEASURED_AT_TOLERANCE_MS), + lte: new Date(segment.end.getTime() + MEASURED_AT_TOLERANCE_MS), + }, + }, + select: { + id: true, + type: true, + source: true, + value: true, + measuredAt: true, + }, + }); + const twin = candidates.find((candidate) => + isSameReadingAcrossSource( + { type, source, value: segment.minutes, measuredAt: segment.end }, + candidate, + ), + ); + if (twin) { + return namedSleepStageResponse(twin.id, "duplicate", { + stage: segment.stage, + source, + crossSource: true, + }); + } + } + + const verdict = await prisma.$transaction((tx) => + reconcileExternalMeasurement( + tx, + { + userId, + type, + value: segment.minutes, + unit: getUnitForType(type), + source, + measuredAt: segment.end, + externalId: segment.externalId, + externalSourceVersion: null, + sleepStage: segment.stage, + deviceType: entry.deviceType ?? null, + }, + { exactExternalMatch: "duplicate" }, + ), + ); + + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + if (verdict.status === "rejected_range") { + // The schema applies the same band, so this is the two gates disagreeing. + return apiError("Value out of plausible range", 422, { + errorCode: "measurement.create.invalid", + }); + } + + const status = + verdict.status === "inserted" + ? "inserted" + : verdict.status === "duplicate" + ? "duplicate" + : "updated"; + + if (status !== "duplicate") { + await auditLog("measurement.create", { + userId, + ipAddress: args.ipAddress, + details: { + type, + measurementId: verdict.row.id, + sleepStage: segment.stage, + status, + }, + }); + invalidateUserMeasurements(userId, { evict: true }); + fireAndForget(enqueueReminderSatisfy(userId), { + action: "reminder.satisfy.enqueue", + }); + if (status === "inserted") { + void emitInsertedMeasurementArrivals( + userId, + [verdict.row], + "manual", + ).catch(() => {}); + void maybeEnqueueMorningRefresh(userId, [segment.end]).catch(() => {}); + } + await afterMeasurementMutation( + userId, + [{ type, measuredAt: segment.end }, ...(verdict.dirtyIdentities ?? [])], + { source: "manual" }, + ); + } + + return namedSleepStageResponse(verdict.row.id, status, { + stage: segment.stage, + source, + crossSource: false, + }); +} + +async function namedSleepStageResponse( + id: string, + status: "inserted" | "updated" | "duplicate", + meta: { stage: string; source: string; crossSource: boolean }, +): Promise { + annotate({ + action: { + name: "measurement.create.sleep-stage", + entity_type: "measurement", + entity_id: id, + }, + meta: { + status, + stage: meta.stage, + source: meta.source, + cross_source_merge: meta.crossSource, + }, + }); + const row = await prisma.measurement.findUniqueOrThrow({ where: { id } }); + return apiSuccess( + { ...shapeMeasurementNotes(row), status }, + status === "inserted" ? 201 : 200, + ); +} diff --git a/src/lib/measurements/list/measurement-list.ts b/src/lib/measurements/list/measurement-list.ts new file mode 100644 index 000000000..57e0535fd --- /dev/null +++ b/src/lib/measurements/list/measurement-list.ts @@ -0,0 +1,311 @@ +/** + * `GET /api/measurements` after its query has been validated: pick the list + * mode the query asks for and answer it. + * + * The modes are tried in a fixed order: the per-night sleep list, the + * cumulative day drill-down, the day-grouped cumulative list, the chart's + * rollup series, a live aggregate, and the plain page. The route owns + * authentication and the query refusal. + */ +import type { z } from "zod/v4"; + +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { apiSuccess } from "@/lib/api-response"; +import type { listMeasurementsSchema } from "@/lib/validations/measurement"; +import { + BUCKET_CAP, + type AggregateGrain, +} from "@/lib/measurements/range-aggregation"; +import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { localDayWindow } from "@/lib/tz/local-day"; +import { shapeMeasurementNotes } from "@/lib/crypto/note-cipher"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { + readDailySeries, + readLiveBuckets, +} from "@/lib/measurements/daily-series-read"; +import { dayGroupedListResponse } from "@/lib/measurements/list/day-grouped"; +import { sleepListResponse } from "@/lib/measurements/list/sleep-list"; +import { resolveStoredTimezone, resolveUserTimezone } from "@/lib/tz/resolver"; +import type { MeasurementType } from "@/generated/prisma/client"; + +type ListQuery = z.output; +type ListUser = { id: string; timezone: string | null }; + +export async function measurementListResponse( + user: ListUser, + query: ListQuery, +): Promise { + const { type, from, to, limit, offset, sortDir, groupBy, dayKey } = query; + const { aggregate, source, sourceEq } = query; + + // Sleep is read row-per-STAGE (≈5 rows/night × source-count), not a spot + // sample and not a cumulative day-sum. A plain list query for + // SLEEP_DURATION would leak every stage segment as its own "value" — the + // "many small numbers that stand for themselves" complaint. Route it + // through the canonical per-night aggregation (the same helper the chart + // series + dashboard tile use) so a sleep list returns one row per night + // carrying the night's TIME ASLEEP, with a `dayKey` drill-down to that + // night's stage segments. Storage is untouched; only the display + // collapses. Runs ahead of the cumulative branches because sleep is in + // neither CUMULATIVE_HK_TYPES nor the spot-sample set. + if (type === "SLEEP_DURATION") { + return sleepListResponse(user, { + from, + to, + limit, + offset, + sortDir, + dayKey, + }); + } + + // List-view drill-down to per-sample rows for a single calendar day in the + // user's IANA timezone. Gated to cumulative HK types — the spot-sample + // types render one-row-per-event already and have nothing to drill into. + if (dayKey && type && CUMULATIVE_HK_TYPES.has(type as MeasurementType)) { + return drillDownResponse(user, type as MeasurementType, query); + } + + // Collapsed "one row per day" mode for cumulative types. The list page + // renders the day's SUM as a single synthesised row with a `sampleCount` + // chevron that drills back into the per-sample rows via the `dayKey=…` + // branch above. `limit` / `offset` page over DAYS, not raw rows (#1245). + if ( + groupBy === "day" && + type && + CUMULATIVE_HK_TYPES.has(type as MeasurementType) + ) { + return dayGroupedListResponse(user, { + type: type as MeasurementType, + limit, + offset, + sortDir, + sourceEq, + from, + to, + }); + } + + // Server-side aggregation. Gated on an explicit `aggregate` param so the + // iOS contract (raw `MeasurementWireDTO` shape on `GET /api/measurements`) + // is byte-stable for any caller that omits the query parameter. The + // chart-data client opts in; iOS does not. + // + // `take` does not apply BEFORE bucketising. The aggregation runs in + // Postgres via `date_trunc`, the bucket cap is applied AFTER, so a 1-year + // `aggregate=daily` window walks every row in the window and returns up to + // 365 buckets per type instead of truncating to the first N raw rows. + // + // "All time" range semantics. When the chart's "All" tab is active, the + // client passes `from` = the user's earliest measurement (or a sentinel + // like 1970-01-01) plus `to` = now plus `aggregate=monthly` (or `weekly` + // when full history < 2 years). The response is bounded by the + // `BUCKET_CAP` ceiling per grain (monthly: 24, weekly: 105, daily: 365) so + // a multi-decade account never paints an unbounded series. + // + // `source=rollup` + `aggregate=daily` — the chart's series read. Windows up + // to the DAY cap fold live into the user's local days (#1026); wider + // windows step up to the WEEK / MONTH / YEAR rollup tier. `type` is + // mandatory so the response mirrors the per-type ask the chart makes. + if (source === "rollup" && aggregate === "daily" && type && from && to) { + const series = await rollupSeriesResponse( + user.id, + type as MeasurementType, + { from, to, limit }, + ); + if (series) return series; + // An empty result means the window genuinely holds no rows. Fall + // through to the shared live aggregate branch below to keep the + // response shape identical for the truly-empty case. + } + + if (aggregate && aggregate !== "raw" && from && to) { + return liveAggregateResponse(user.id, query, aggregate, { from, to }); + } + + return plainPageResponse(user.id, query); +} + +async function drillDownResponse( + user: ListUser, + type: MeasurementType, + query: ListQuery, +): Promise { + const { limit, sortDir } = query; + const dayKey = query.dayKey as string; + // The same zone the day-grouped list cuts its days in, so a row's + // drill-down opens exactly the day the row summed. + const tz = await resolveStoredTimezone(user.timezone); + // Resolve `[dayStart, dayEnd)` for the requested calendar day in the + // user's local zone. `localDayWindow` honours DST: the earlier shape + // (`canonicalDailyTimestamp ± 12 h`) silently leaked or hid one hour of + // samples on spring-forward (23-h day) and fall-back (25-h day) by always + // spanning exactly 24 h. Works for every IANA offset including + // sub-half-hour zones (Asia/Kathmandu UTC+5:45, Pacific/Chatham UTC+12:45). + const { dayStart, dayEnd } = localDayWindow(dayKey, tz); + const samples = await prisma.measurement.findMany({ + where: { + userId: user.id, + type, + measuredAt: { gte: dayStart, lt: dayEnd }, + deletedAt: null, + }, + // Same tiebreaker as the paged list. This one slices in memory rather + // than with `skip`, so there is no page drift to prevent — but two + // identical requests would still be free to render a tied block in + // different orders, and a drill-down that reshuffles on refresh reads + // as data changing. + orderBy: [{ measuredAt: sortDir }, { id: sortDir }], + // The 1000-row cap lives on the validator (a refine on `(limit, dayKey)` + // returning 422 when a caller asks for more). `limit` is read straight + // through because the validator has already clamped it. + take: limit, + }); + annotate({ + action: { name: "measurement.list" }, + meta: { total: samples.length, type, dayKey, mode: "drill-down" }, + }); + return apiSuccess({ + measurements: samples, + meta: { total: samples.length, limit, offset: 0, dayKey }, + }); +} + +/** The chart's daily series, or null when the window holds no rows. */ +async function rollupSeriesResponse( + userId: string, + type: MeasurementType, + window: { from: Date; to: Date; limit: number }, +): Promise { + const { from, to, limit } = window; + // Shared with the batched series endpoint, so both read the same rows. + const cap = Math.min(limit, BUCKET_CAP.daily); + const priorityJson = await loadUserSourcePriority(userId); + const measurements = await readDailySeries({ + userId, + type, + from, + to, + limit, + priorityJson, + }); + if (measurements.length === 0) return null; + return apiSuccess({ + measurements, + meta: { + total: measurements.length, + limit: cap, + offset: 0, + aggregate: "daily", + }, + }); +} + +async function liveAggregateResponse( + userId: string, + query: ListQuery, + aggregate: Exclude, + window: { from: Date; to: Date }, +): Promise { + const { type, limit } = query; + const grain: Exclude = aggregate; + const cap = Math.min(limit, BUCKET_CAP[grain]); + // Buckets are cut at the user's own calendar boundaries (#1026), the + // same day the raw list and the chart's short range use. Cumulative + // types SUM per bucket, every other type averages; overlapping sources + // collapse to the ladder-canonical one first. + const priorityJson = await loadUserSourcePriority(userId); + const measurements = await readLiveBuckets({ + userId, + type: (type as MeasurementType | undefined) ?? null, + from: window.from, + to: window.to, + cap, + priorityJson, + grain, + timeZone: await resolveUserTimezone(userId), + }); + annotate({ + action: { name: "measurement.list" }, + meta: { total: measurements.length, type, aggregate: grain }, + }); + return apiSuccess({ + measurements, + meta: { + total: measurements.length, + limit: cap, + offset: 0, + aggregate: grain, + }, + }); +} + +async function plainPageResponse( + userId: string, + query: ListQuery, +): Promise { + const { type, from, to, limit, offset, sortBy, sortDir } = query; + const { sourceEq, valueMin, valueMax } = query; + const where = { + userId, + // Soft-deleted readings are invisible to every list / count / aggregate + // read. The dedicated `sync/state` route keeps surfacing them via its + // own filter so iOS can reconcile tombstones; every other consumer (list + // view, analytics fan-out, dashboard read paths) must never include them. + deletedAt: null, + ...(type && { type: type as MeasurementType }), + // Management-list source filter. Distinct from the `source=rollup` + // opt-in (which selects the rollup read tier); `sourceEq` narrows the + // plain list to one ingest source. + ...(sourceEq && { source: sourceEq }), + // Value-range filter. Open-ended on either side; the validator has + // already rejected an inverted `valueMin > valueMax`. + ...(valueMin != null || valueMax != null + ? { + value: { + ...(valueMin != null && { gte: valueMin }), + ...(valueMax != null && { lte: valueMax }), + }, + } + : {}), + ...(from || to + ? { + measuredAt: { + ...(from && { gte: from }), + ...(to && { lte: to }), + }, + } + : {}), + }; + + const [measurements, total] = await Promise.all([ + prisma.measurement.findMany({ + where, + // `{ id: sortDir }` is the tiebreaker, not decoration. `measuredAt` + // is not unique here and ties are the normal case. Only the exact + // tuple `(userId, type, measuredAt, source, sleepStage)` is unique, + // so a tie survives across `type` (the `stats::YYYY-MM-DD` + // daily-rollup convention puts every daily total at one instant), + // across `source` (two writers landing on the same instant) and + // across `sleepStage`. Without a unique secondary key Postgres is + // free to order the tied block differently between the query for page + // N and the query for page N+1, so a client paging its own history + // silently gets one row twice and never gets another. + orderBy: [{ [sortBy]: sortDir }, { id: sortDir }], + take: limit, + skip: offset, + }), + prisma.measurement.count({ where }), + ]); + + annotate({ action: { name: "measurement.list" }, meta: { total, type } }); + + return apiSuccess({ + // Decrypt `notesEncrypted` onto `notes` and strip the ciphertext before + // it leaves the server (iOS MeasurementWireDTO contract). + measurements: measurements.map(shapeMeasurementNotes), + meta: { total, limit, offset }, + }); +}