From 24af254249127195c65ca79f985b400899738abd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 23 Oct 2026 14:47:57 +0200 Subject: [PATCH] refactor(medications): move the bulk dose ingest out of its route POST /api/medications/intake/bulk keeps authentication, its own bulk bucket, the body with its size and schema refusals, the payload schema and the HTTP mapping. The ingest, one 846-line function, moves to lib/medications/intake-bulk/ in the order it always ran: the ownership read, the Apple mirror gate and the batch pre-reads (ingest.ts), each entry's refusals, site, write bookkeeping and P2002 answer (entry.ts), its slot attribution (entry-attribution.ts) and its slot, replay, probe or standalone write (entry-write.ts), then the audit row, the dose days, the snooze, the tails, the arrival record and the checkpoint. The two identical canonical-slot write blocks of the loop are one function now. The route falls from 1044 to 231 lines; no function in the new modules is over 70. The parity walk recorded before the move (medication-intake-bulk.parity) is unchanged. The per-entry external-id markers name the module that classifies each entry, the batch-surface and key-scope legs read the moved code, and the intake-tracking allowlist and the dense-read guard follow it. --- .../dense-measurement-read-guard.test.ts | 1 + .../external-id-stability-guard.test.ts | 7 +- .../medication-intake-tracking-guard.test.ts | 2 +- src/app/api/medications/intake/bulk/route.ts | 847 +----------------- .../medications/intake-bulk/batch-state.ts | 102 +++ .../intake-bulk/entry-attribution.ts | 178 ++++ .../medications/intake-bulk/entry-write.ts | 251 ++++++ src/lib/medications/intake-bulk/entry.ts | 303 +++++++ src/lib/medications/intake-bulk/ingest.ts | 321 +++++++ 9 files changed, 1179 insertions(+), 833 deletions(-) create mode 100644 src/lib/medications/intake-bulk/batch-state.ts create mode 100644 src/lib/medications/intake-bulk/entry-attribution.ts create mode 100644 src/lib/medications/intake-bulk/entry-write.ts create mode 100644 src/lib/medications/intake-bulk/entry.ts create mode 100644 src/lib/medications/intake-bulk/ingest.ts diff --git a/src/__tests__/dense-measurement-read-guard.test.ts b/src/__tests__/dense-measurement-read-guard.test.ts index 497c146d6..4a3f5ed3e 100644 --- a/src/__tests__/dense-measurement-read-guard.test.ts +++ b/src/__tests__/dense-measurement-read-guard.test.ts @@ -77,6 +77,7 @@ const SCOPED_DIRS = [ "lib/medications/intake-today.ts", "lib/medications/intake-compliance.ts", "lib/medications/update-intake-status/", + "lib/medications/intake-bulk/", ]; /** Types a device writes at sampling rate rather than a few times a day. */ diff --git a/src/__tests__/external-id-stability-guard.test.ts b/src/__tests__/external-id-stability-guard.test.ts index 75b8eb974..5d84f65e3 100644 --- a/src/__tests__/external-id-stability-guard.test.ts +++ b/src/__tests__/external-id-stability-guard.test.ts @@ -365,7 +365,7 @@ describe("the intake idempotency key is never read across users", () => { const found = intakeQueriesFilteringOnTheKey(); expect(found.length).toBeGreaterThanOrEqual(6); expect(found.map((f) => f.file)).toContain( - "app/api/medications/intake/bulk/route.ts", + "lib/medications/intake-bulk/entry-write.ts", ); }); @@ -403,7 +403,10 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => { "lib/measurements/batch-ingest/ingest.ts", ], "app/api/mood-entries/bulk/route.ts": [], - "app/api/medications/intake/bulk/route.ts": [], + "app/api/medications/intake/bulk/route.ts": [ + "lib/medications/intake-bulk/entry.ts", + "lib/medications/intake-bulk/ingest.ts", + ], "app/api/workouts/batch/route.ts": [ "lib/workouts/batch-ingest/entries.ts", "lib/workouts/batch-ingest/ingest.ts", diff --git a/src/__tests__/medication-intake-tracking-guard.test.ts b/src/__tests__/medication-intake-tracking-guard.test.ts index 3df0ec79d..fdd580c62 100644 --- a/src/__tests__/medication-intake-tracking-guard.test.ts +++ b/src/__tests__/medication-intake-tracking-guard.test.ts @@ -72,7 +72,7 @@ const INTAKE_MAY_SEE_EVERY_ROW: Record = { "deletes rows the person chose", "lib/medications/update-medication/write.ts": "one medication's open slots, tombstoned on a schedule replace or the tracking switch", - "app/api/medications/intake/bulk/route.ts": "writes recorded doses", + "lib/medications/intake-bulk/ingest.ts": "writes recorded doses", "lib/medications/scheduling/slot-upsert.ts": "binds a recorded dose to its slot", "lib/medications/scheduling/worker-helpers.ts": diff --git a/src/app/api/medications/intake/bulk/route.ts b/src/app/api/medications/intake/bulk/route.ts index 7e6f9a453..80d490cb6 100644 --- a/src/app/api/medications/intake/bulk/route.ts +++ b/src/app/api/medications/intake/bulk/route.ts @@ -45,7 +45,6 @@ import { z } from "zod/v4"; import { prisma } from "@/lib/db"; import { apiHandler, isScopedCredential, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; -import { auditLog } from "@/lib/auth/audit"; import { apiError, apiSuccess, @@ -56,52 +55,13 @@ import { } from "@/lib/api-response"; import { withIdempotency } from "@/lib/idempotency"; import { checkRateLimit } from "@/lib/rate-limit"; -import { markSyncCheckpoint } from "@/lib/sync/checkpoint"; -import { - BACKFILL_SYNC_TRIGGER, - HEALTHKIT_REQUEST_SYNC_TRIGGERS, - MEDICATION_DOSE_ARRIVAL_TYPE, - recordHealthKitTypeArrivals, -} from "@/lib/integrations/healthkit-type-sync"; -import { invalidateUserMedications } from "@/lib/cache/invalidate"; -import { - dayKeyForScheduledFor, - refreshDoseDay, -} from "@/lib/medications/dose-day-refresh"; -import { - applyCanonicalSlotWrite, - findPinConflict, - mayConvergeOntoSuppliedSlot, - resolveForcedSlotForWrite, - resolveSlotForWriteByBand, - resolveSlotInstantForWrite, - slotHeldByAnotherDose, -} from "@/lib/medications/scheduling/slot-upsert"; -import { resolveInjectionSiteForWrite } from "@/lib/medications/injection-site-write"; -import { - consumeForIntake, - restoreForIntake, -} from "@/lib/medications/inventory/consumption"; +import { HEALTHKIT_REQUEST_SYNC_TRIGGERS } from "@/lib/integrations/healthkit-type-sync"; import { boundedTakenAtSchema, injectionSiteEnum, - type InjectionSiteValue, } from "@/lib/validations/medication"; -import { - classifyExternalId, - unstableExternalIdMeta, - UNSTABLE_EXTERNAL_ID_REASON, - type UnstableExternalIdShape, -} from "@/lib/validations/external-id"; -import type { InjectionSiteKey } from "@/lib/medications/injection-sites"; -import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync"; -import { dispatchMedicationIntakeWebClearBulk } from "@/lib/notifications/web-push-clear"; -import { countOutstandingDosesToday } from "@/lib/medications/outstanding-doses"; -import { - INTAKE_NOT_TRACKED_REASON, - isRecordOnly, -} from "@/lib/medications/intake-tracking"; import { isoDateTime } from "@/lib/validations/datetime"; +import { ingestBulkIntake } from "@/lib/medications/intake-bulk/ingest"; const MAX_ENTRIES_PER_BATCH = 500; const BATCH_RATE_LIMIT_MAX = 60; @@ -121,7 +81,7 @@ const bulkEntrySchema = z // equality forever, so it is an identity and carries the same // stability floor. Checked per entry for the same reason as // `externalId` above. - // @external-id-checked-per-entry: src/app/api/medications/intake/bulk/route.ts + // @external-id-checked-per-entry: src/lib/medications/intake-bulk/entry.ts idempotencyKey: z.string().min(1).max(128).optional(), // v1.8.5 — optional per-entry injection site. Validated against the // medication's effective allowed set; a disallowed site marks the @@ -157,7 +117,7 @@ const bulkEntrySchema = z // 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/medications/intake/bulk/route.ts + // @external-id-checked-per-entry: src/lib/medications/intake-bulk/entry.ts externalId: z.string().trim().min(1).max(128).optional(), }) .refine( @@ -181,19 +141,6 @@ const bulkPayloadSchema = z.object({ syncTrigger: z.enum(HEALTHKIT_REQUEST_SYNC_TRIGGERS).optional(), }); -// v1.8.2 — `updated` joins the per-entry status vocabulary: a write that -// snaps onto an existing scheduled-slot row (e.g. the pending REMINDER -// row) updates it in place rather than inserting. The iOS sync engine -// treats `updated` the same as `inserted` for cursor advancement — both -// mean "the server accepted this entry and produced a row id". -type EntryStatus = "inserted" | "updated" | "duplicate" | "skipped"; -interface EntryResult { - index: number; - status: EntryStatus; - reason?: string; - id?: string; -} - export const POST = apiHandler(withIdempotency<[NextRequest]>(postBulk)); async function postBulk(request: NextRequest): Promise { @@ -203,8 +150,8 @@ async function postBulk(request: NextRequest): Promise { const auth = await requireAuth(undefined, { writeBucket: "own" }); const { user } = auth; // A narrow, single-purpose credential (a bridge, an ingest token) is no - // Apple Health sync, so it never writes the dose arrival record below. The - // bare `requireAuth()` refuses such a credential today; the check keeps the + // Apple Health sync, so it never writes the dose arrival record. The bare + // `requireAuth()` refuses such a credential today; the check keeps the // record honest should this route ever declare a scope. const scoped = isScopedCredential(auth); @@ -238,7 +185,7 @@ async function postBulk(request: NextRequest): Promise { const parsed = bulkPayloadSchema.safeParse(rawBody); if (!parsed.success) { - // v1.4.43 — bulk intake ingest; preserve the existing + // Bulk intake ingest; preserve the existing // `medications.intake.bulk.invalid` errorCode meta so the iOS // Sync engine's retry classifier still branches on it. Audit // breadcrumb keyed `medications.intake.bulk.validation-failed`. @@ -247,7 +194,7 @@ async function postBulk(request: NextRequest): Promise { action: { name: "medications.intake.bulk.validation-failed" }, meta: { issue_count: issues.length }, }); - // v1.4.49 — strip `message` from the audit-ledger row. + // Strip `message` from the audit-ledger row. const auditIssues = sanitiseZodIssues(parsed.error.issues, { stripValuesFromMessage: true, }); @@ -269,776 +216,16 @@ async function postBulk(request: NextRequest): Promise { const { entries, syncTrigger } = parsed.data; - // Up-front ownership check — the user can only POST intake events - // against medications they own. One read covers every medicationId - // in the batch. - const medicationIds = Array.from(new Set(entries.map((e) => e.medicationId))); - const ownedMedications = await prisma.medication.findMany({ - where: { id: { in: medicationIds }, userId: user.id }, - // v1.8.5 — pull the injection-tracking fields so a per-entry - // `injectionSite` can be resolved + validated without an extra - // per-entry read. v1.28 — `externalSource` rides along so the - // Apple-entry mirror gate below needs no extra read either. - select: { - id: true, - deliveryForm: true, - trackInjectionSites: true, - allowedInjectionSites: true, - externalSource: true, - trackIntake: true, - }, - }); - const ownedSet = new Set(ownedMedications.map((m) => m.id)); - const medById = new Map(ownedMedications.map((m) => [m.id, m])); - - // v1.28 — source-exclusive Apple-mirrored medications: an APPLE_HEALTH - // dose event may only attach to a medication mirrored from Apple Health - // (`Medication.externalSource === "APPLE_HEALTH"`). Targeting a native - // medication is a client contract violation (the iOS sync maps HK dose - // events onto the meds it mirrored itself), not a per-entry data - // problem, so the whole batch 422s with a stable errorCode the client - // can branch on. Unowned medicationIds keep the per-entry - // `medication_not_found` skip below. - const appleMismatch = entries.some( - (e) => - e.source === "APPLE_HEALTH" && - ownedSet.has(e.medicationId) && - medById.get(e.medicationId)?.externalSource !== "APPLE_HEALTH", - ); - if (appleMismatch) { - return apiError( - "An APPLE_HEALTH entry may only target a medication mirrored from Apple Health", - 422, - { errorCode: "medications.intake.bulk.apple_health_not_mirrored" }, - ); - } - - // v1.28 — idempotent re-sync dedup: one read collects every live row - // already carrying one of the batch's external dose-event UUIDs; the - // per-entry loop reports those entries as `duplicate` (first-write-wins - // per Apple dose) without touching the DB again. Fresh inserts register - // into the same map so an intra-batch repeat of a UUID dedups too. The - // partial `(user_id, external_id)` unique (migration 0239) is the race - // backstop. - const batchExternalIds = Array.from( - new Set( - entries - .map((e) => e.externalId) - .filter((id): id is string => id !== undefined), - ), - ); - const rowIdByExternalId = new Map(); - if (batchExternalIds.length > 0) { - const existingExternalRows = await prisma.medicationIntakeEvent.findMany({ - where: { - userId: user.id, - externalId: { in: batchExternalIds }, - deletedAt: null, - }, - select: { id: true, externalId: true }, - }); - for (const row of existingExternalRows) { - if (row.externalId) rowIdByExternalId.set(row.externalId, row.id); - } - } - - // v1.8.5 — the user's global exclusion deny-list, loaded once for the - // whole batch (it is user-scoped, not per-medication). Only read when - // at least one entry actually carries an `injectionSite`, so the hot - // sync path (no site) pays for zero extra round-trips. - const batchHasInjectionSite = entries.some( - (e) => e.injectionSite !== undefined, - ); - let globalExcluded: InjectionSiteValue[] = []; - if (batchHasInjectionSite) { - const userRow = await prisma.user.findUnique({ - where: { id: user.id }, - select: { globalExcludedInjectionSites: true }, - }); - globalExcluded = (userRow?.globalExcludedInjectionSites ?? - []) as InjectionSiteValue[]; - } - - const results: EntryResult[] = []; - let inserted = 0; - let updated = 0; - let duplicates = 0; - // External dose records kept as their own unscheduled dose because their - // band slot already carried a different one. - let extraDosesUnscheduled = 0; - const skipped: Array<{ index: number; reason: string }> = []; - const unstableShapes: UnstableExternalIdShape[] = []; - // Distinct `(medicationId, dayKey)` pairs touched by the batch, so one - // dose-day refresh fires per pair after all inserts complete. A per-row - // refresh would balloon a 500-entry batch into 500 sequential refreshes. - const touchedDays = new Map< - string, - { medicationId: string; dayKey: string } - >(); - // v1.17.1 (#22) — distinct `(medicationId, scheduledFor)` slots whose - // state actually changed in this batch (a row inserted / updated), so the - // silent cross-device sync fires once per affected slot — not once per - // row, and not for pending echoes / duplicates that changed nothing. - const syncSlots = new Map< - string, - { medicationId: string; scheduledFor: string } - >(); - // A landed non-skipped dose quiets an active snooze, exactly like the - // single-intake routes (they null `snoozedUntil` on every recorded take). - // Collected per medication here, cleared once after the loop. - const snoozeClearIds = new Set(); - - for (let i = 0; i < entries.length; i++) { - const entry = entries[i]; - - if (!ownedSet.has(entry.medicationId)) { - const reason = "medication_not_found"; - skipped.push({ index: i, reason }); - results.push({ index: i, status: "skipped", reason }); - continue; - } - - // A record-only medication takes no new dose (iOS #116, item 15). The - // single-dose routes answer 422 `medication.intake.notTracked`; here the - // same decision is one entry's outcome, so the rest of the batch lands. - if (isRecordOnly(medById.get(entry.medicationId) ?? {})) { - const reason = INTAKE_NOT_TRACKED_REASON; - skipped.push({ index: i, reason }); - results.push({ index: i, status: "skipped", reason }); - continue; - } - - // Both dedup keys on this route are only idempotent while the value - // is STABLE across client launches: `(userId, externalId)` for the - // Apple dose event, and the `@unique` `idempotency_key` column for - // the replay probe below. A value that rotates per process (an - // object description carrying a memory address) never matches its - // own earlier row, so every re-sync logs the dose again. Refused per - // entry — one bad row must not stop the rest of the batch landing. - const unstable = - (entry.externalId === undefined - ? null - : classifyExternalId(entry.externalId)) ?? - (entry.idempotencyKey === undefined - ? null - : classifyExternalId(entry.idempotencyKey)); - if (unstable) { - unstableShapes.push(unstable); - skipped.push({ index: i, reason: UNSTABLE_EXTERNAL_ID_REASON }); - results.push({ - index: i, - status: "skipped", - reason: UNSTABLE_EXTERNAL_ID_REASON, - }); - continue; - } - - // v1.28 — Apple dose-event replay: the batch pre-check found a live - // row already carrying this external UUID (or an earlier entry in - // THIS batch inserted it). First-write-wins per Apple dose — report - // `duplicate` with the existing row id so the sync cursor advances. - if (entry.externalId !== undefined) { - const existingRowId = rowIdByExternalId.get(entry.externalId); - if (existingRowId !== undefined) { - duplicates += 1; - results.push({ index: i, status: "duplicate", id: existingRowId }); - continue; - } - } - - try { - const incomingScheduledFor = - entry.scheduledFor ?? entry.takenAt ?? new Date(); - - // Classify the incoming write. An offline-sync replay echoes a - // PENDING projection (no `takenAt`, `skipped:false`) for a slot the - // user may already have actioned; that echo must NEVER clear a - // recorded `takenAt`. An explicit `takenAt` or `skipped:true` is a - // real user action and applies last-write-wins. - const isExplicitTaken = !entry.skipped && entry.takenAt !== undefined; - const isExplicitSkip = entry.skipped === true; - - // v1.15.20 — taken writes attribute through the window-band engine - // (the SAME minter the read ledger + the compliance % consume), so a - // bulk-synced take binds to a slot identically to the single-route - // write. The optional `forceSlotInstant` pins an off-window take onto - // a named real slot. Pending echoes and skips carry no take to - // attribute by, so they keep the canonical anchor snap that binds - // them to the projector/worker-minted slot row. - let canonicalSlot: Date | null; - // v1.15.20 — binding provenance: USER_PIN on the forced path, AUTO on - // a band decision. Pending echoes / skips carry no decision - // (undefined → the upsert leaves an existing USER_PIN untouched). - let attributionSource: "AUTO" | "USER_PIN" | undefined; - // Set when the band slot is held by a different dose and this take is - // recorded unscheduled; it must not converge onto any slot row below. - let heldByAnotherDose = false; - if (isExplicitTaken && entry.forceSlotInstant !== undefined) { - canonicalSlot = await resolveForcedSlotForWrite({ - userId: user.id, - medicationId: entry.medicationId, - userTz: user.timezone, - slotInstant: entry.forceSlotInstant, - }); - if (canonicalSlot === null) { - // Mirrors the single route's 422 (`medications.intake.force_slot - // .invalid`) as a per-entry skip so one bad pin never fails the - // batch. - const reason = "force_slot_invalid"; - skipped.push({ index: i, reason }); - results.push({ index: i, status: "skipped", reason }); - continue; - } - // v1.16.0 — refuse to pin onto a slot another recorded action - // already serves: the explicit-write last-write-wins rule would - // silently overwrite that dose record. Per-entry skip, mirroring - // the single route's 422 `medications.intake.force_slot.occupied`. - if ( - await findPinConflict({ - userId: user.id, - medicationId: entry.medicationId, - canonicalSlot, - incomingTakenAt: entry.takenAt ?? null, - }) - ) { - const reason = "force_slot_occupied"; - skipped.push({ index: i, reason }); - results.push({ index: i, status: "skipped", reason }); - continue; - } - attributionSource = "USER_PIN"; - } else if (isExplicitTaken) { - const attribution = await resolveSlotForWriteByBand({ - userId: user.id, - medicationId: entry.medicationId, - userTz: user.timezone, - takenAt: entry.takenAt as Date, - }); - canonicalSlot = attribution.slotInstant; - attributionSource = "AUTO"; - // An external dose record (a HealthKit dose event) whose band slot - // already carries a different recorded dose is an extra dose, not a - // correction of that one. Converging would overwrite the slot's - // take last-write-wins and the extra dose would exist nowhere; it is - // recorded as its own unscheduled dose instead. - if ( - canonicalSlot !== null && - entry.externalId !== undefined && - (await slotHeldByAnotherDose({ - userId: user.id, - medicationId: entry.medicationId, - canonicalSlot, - takenAt: entry.takenAt as Date, - externalId: entry.externalId, - scheduledFor: entry.scheduledFor ?? null, - userTz: user.timezone, - })) - ) { - canonicalSlot = null; - heldByAnotherDose = true; - extraDosesUnscheduled += 1; - } - } else if (isExplicitSkip && entry.scheduledFor === undefined) { - // v1.16.9 — a slot-less deliberate skip resolves by band - // membership on the skip moment, exactly like a take does. The - // defaulted-now snap below only reaches the tight reminder-grace - // window, so a skip tapped 20 minutes past the dose time inserted - // an orphan ad-hoc skip row while the slot's pending REMINDER row - // stayed open and later auto-missed — the user's deliberate skip - // recorded as a miss. - const attribution = await resolveSlotForWriteByBand({ - userId: user.id, - medicationId: entry.medicationId, - userTz: user.timezone, - takenAt: incomingScheduledFor, - }); - canonicalSlot = attribution.slotInstant; - } else { - // v1.8.2 — source-agnostic anchor snap for pending echoes + skips. - // iOS posts the reminder actions here (source API); without this - // they inserted a SECOND row for a slot already carrying the - // projector/worker's pending REMINDER row (the unique key includes - // `source`, and the iOS-vs-server `scheduledFor` drifts by a - // minute). Snap onto the canonical slot instant and update the - // existing row in place. - canonicalSlot = await resolveSlotInstantForWrite({ - userId: user.id, - medicationId: entry.medicationId, - userTz: user.timezone, - incoming: incomingScheduledFor, - // Only a client-supplied `scheduledFor` names a real slot. A - // defaulted-now write must not snap across the wide ±halfGap - // window onto a far slot (phantom morning dose). - instantIsExplicit: entry.scheduledFor !== undefined, - // Pending echoes / skips never record a take; a pending sync - // echo legitimately maps to a future slot, so the taken - // future-slot guard stays off here. - isTakenWrite: false, - }); - } - - // The `scheduledFor` instant the row actually lands on — set by the - // branch that performs the write so the rollup recompute below keys - // the same day the stored row anchors to. - let effectiveScheduledFor = canonicalSlot ?? incomingScheduledFor; - - // v1.8.5 — resolve + validate the optional per-entry injection - // site. A disallowed site marks the entry skipped without failing - // the batch (per-entry contract); a non-injection / tracking-off / - // non-taken entry silently drops it. - let resolvedInjectionSite: InjectionSiteKey | null = null; - if (entry.injectionSite !== undefined) { - const med = medById.get(entry.medicationId); - const resolution = resolveInjectionSiteForWrite({ - submitted: entry.injectionSite, - taken: isExplicitTaken, - deliveryForm: med?.deliveryForm ?? "ORAL", - trackInjectionSites: med?.trackInjectionSites ?? false, - allowedInjectionSites: (med?.allowedInjectionSites ?? - []) as InjectionSiteKey[], - globalExcludedInjectionSites: globalExcluded as InjectionSiteKey[], - }); - if (resolution.kind === "disallowed") { - const reason = "injection_site_not_allowed"; - skipped.push({ index: i, reason }); - results.push({ index: i, status: "skipped", reason }); - continue; - } - resolvedInjectionSite = resolution.site; - } - - if (canonicalSlot) { - // Scheduled dose — converge onto the one canonical slot row through - // the shared upsert: H1 deterministic selection, C2 no-downgrade - // guard, and a C1 race-safe create that re-finds + updates on a - // P2002 collision rather than misclassifying it as a duplicate and - // dropping the dose. - const applied = await applyCanonicalSlotWrite({ - client: prisma, - userId: user.id, - medicationId: entry.medicationId, - canonicalSlot, - takenAt: entry.takenAt ?? null, - skipped: entry.skipped, - isExplicitTaken, - isExplicitSkip, - idempotencyKey: entry.idempotencyKey ?? null, - // v1.28 — an Apple dose-event import stamps its provenance on a - // fresh row; every other caller keeps today's "API". - createSource: entry.source ?? "API", - injectionSite: resolvedInjectionSite, - attributionSource, - // v1.16.4 — dose override only documents a consumed dose. - doseTaken: (!entry.skipped && entry.doseTaken) || null, - // v1.28 — external dose-event UUID rides onto the landed row. - externalId: entry.externalId ?? null, - }); - if (applied.noDowngradeNoOp) { - // Pending echo onto an already-actioned slot. Report it as a - // duplicate so the iOS cursor advances WITHOUT downgrading the - // recorded dose. A pending echo is never a taken write, so it - // must never reach the inventory consume below. - duplicates += 1; - results.push({ index: i, status: "duplicate", id: applied.row.id }); - } else if (applied.outcome === "updated") { - updated += 1; - results.push({ index: i, status: "updated", id: applied.row.id }); - } else { - inserted += 1; - results.push({ index: i, status: "inserted", id: applied.row.id }); - } - // v1.16.10 — inventory: only the genuine pending→taken - // transition consumes (the upsert's `consumedTransition` flag — - // the ONLY transition that may move stock). The stamp keeps a - // replayed batch exactly-once for post-v1.16.10 rows; the - // transition gate additionally protects pre-stamp rows: a full - // journal replay re-posting historical taken doses must not - // drain today's stock through their NULL stamps. An explicit - // skip refunds a previously-taken row's stamp. Pending echoes - // carry neither flag and never touch stock. - if (applied.consumedTransition && isExplicitTaken) { - await consumeForIntake({ - client: prisma, - userId: user.id, - medicationId: entry.medicationId, - eventId: applied.row.id, - intakeAt: entry.takenAt as Date, - }); - } else if (!applied.noDowngradeNoOp && isExplicitSkip) { - await restoreForIntake({ - client: prisma, - userId: user.id, - eventId: applied.row.id, - }); - } - } else { - // Resolver-null: PRN / off-slot / future-slot-guarded. Before the - // standalone insert, two guards keep the slot from forking into a - // second live row: - // - // 1. idempotencyKey replay pre-check — a re-submission returns the - // existing row as "duplicate" (previously reached via the P2002 - // catch; the explicit probe keeps that contract now that a - // replay could otherwise converge as "updated" below). - // 2. source-agnostic convergence probe — when ANY live row already - // sits on the incoming instant (e.g. the pending REMINDER row - // the worker pre-minted on a slot the taken-write future guard - // refused to snap to), converge onto THAT row through the shared - // slot upsert instead of inserting a sibling that differs only - // by `source`. The partial unique index carries `source`, so it - // cannot catch this duplicate; the probe must. - if (entry.idempotencyKey) { - // Scoped to the caller. The key is client-supplied, so an - // unscoped read would resolve another user's row and hand its - // id back here (migration 0273 made the unique per user). - const replay = await prisma.medicationIntakeEvent.findUnique({ - where: { - userId_idempotencyKey: { - userId: user.id, - idempotencyKey: entry.idempotencyKey, - }, - }, - select: { id: true }, - }); - if (replay) { - duplicates += 1; - results.push({ index: i, status: "duplicate", id: replay.id }); - continue; - } - } - // Only an explicit client `scheduledFor` names a slot a row could - // already sit on; a defaulted anchor (takenAt / now) never does, so - // the probe is skipped on that path. - // - // Dose-safety guard: a TAKEN entry must not converge onto a slot - // whose anchor is in the future relative to the take. The resolver's - // forward guard already refused to snap a late take onto a future - // slot; the probe must honour the same invariant rather than binding - // the take forward onto that slot's pending REMINDER row (a - // late-morning dose consuming the evening slot). It records - // standalone (ad-hoc) instead. - const existingSlotRow = - !heldByAnotherDose && - entry.scheduledFor !== undefined && - mayConvergeOntoSuppliedSlot({ - skipped: entry.skipped, - takenAt: entry.takenAt ?? null, - suppliedSlot: incomingScheduledFor, - }) - ? await prisma.medicationIntakeEvent.findFirst({ - where: { - userId: user.id, - medicationId: entry.medicationId, - scheduledFor: incomingScheduledFor, - deletedAt: null, - }, - select: { id: true }, - }) - : null; - if (existingSlotRow) { - const applied = await applyCanonicalSlotWrite({ - client: prisma, - userId: user.id, - medicationId: entry.medicationId, - canonicalSlot: incomingScheduledFor, - takenAt: entry.takenAt ?? null, - skipped: entry.skipped, - isExplicitTaken, - isExplicitSkip, - idempotencyKey: entry.idempotencyKey ?? null, - // v1.28 — Apple provenance on a fresh row (see above). - createSource: entry.source ?? "API", - injectionSite: resolvedInjectionSite, - attributionSource, - doseTaken: (!entry.skipped && entry.doseTaken) || null, - // v1.28 — external dose-event UUID rides onto the landed row. - externalId: entry.externalId ?? null, - }); - if (applied.noDowngradeNoOp) { - duplicates += 1; - results.push({ index: i, status: "duplicate", id: applied.row.id }); - } else if (applied.outcome === "updated") { - updated += 1; - results.push({ index: i, status: "updated", id: applied.row.id }); - } else { - inserted += 1; - results.push({ index: i, status: "inserted", id: applied.row.id }); - } - // v1.16.10 — same inventory rule as the canonical-slot branch: - // gate on the upsert's pending→taken transition flag. - if (applied.consumedTransition && isExplicitTaken) { - await consumeForIntake({ - client: prisma, - userId: user.id, - medicationId: entry.medicationId, - eventId: applied.row.id, - intakeAt: entry.takenAt as Date, - }); - } else if (!applied.noDowngradeNoOp && isExplicitSkip) { - await restoreForIntake({ - client: prisma, - userId: user.id, - eventId: applied.row.id, - }); - } - } else { - // Genuinely standalone. Anchor the row on the intake instant — - // the documented ad-hoc contract (`scheduledFor = takenAt`) — so - // an unresolvable client anchor (a future slot instant the taken - // guard rejected) can never park a live row exactly on a slot the - // worker later mints a pending REMINDER row for. A pending echo - // (no takenAt) keeps the incoming instant. The idempotencyKey, - // when supplied, has a UNIQUE index; a racing re-submission - // returns the existing row via P2002 → status: "duplicate". - effectiveScheduledFor = entry.takenAt ?? incomingScheduledFor; - const row = await prisma.medicationIntakeEvent.create({ - data: { - userId: user.id, - medicationId: entry.medicationId, - scheduledFor: effectiveScheduledFor, - takenAt: entry.takenAt ?? null, - skipped: entry.skipped, - // v1.28 — Apple provenance on a standalone (ad-hoc) row; - // every other caller keeps today's "API". - source: entry.source ?? "API", - idempotencyKey: entry.idempotencyKey ?? null, - // v1.28 — external dose-event UUID for the idempotent - // re-sync dedup. - ...(entry.externalId !== undefined && { - externalId: entry.externalId, - }), - // v1.8.5 — site only on a resolved taken-injection entry. - ...(resolvedInjectionSite !== null && { - injectionSite: resolvedInjectionSite, - }), - // v1.16.4 — dose override only on a taken entry carrying one. - ...(!entry.skipped && - entry.doseTaken && { doseTaken: entry.doseTaken }), - }, - }); - inserted += 1; - results.push({ index: i, status: "inserted", id: row.id }); - // v1.16.10 — a fresh standalone taken row consumes inventory - // units (a fresh skip / pending row has nothing to consume or - // refund). - if (isExplicitTaken) { - await consumeForIntake({ - client: prisma, - userId: user.id, - medicationId: entry.medicationId, - eventId: row.id, - intakeAt: entry.takenAt as Date, - }); - } - } - } - const dayKey = dayKeyForScheduledFor( - effectiveScheduledFor, - user.timezone, - ); - touchedDays.set(`${entry.medicationId}|${dayKey}`, { - medicationId: entry.medicationId, - dayKey, - }); - // v1.17.1 (#22) — collect the slot for the silent cross-device sync - // only when the row's state actually changed (inserted / updated). A - // pending echo, duplicate, or no-downgrade no-op leaves the visible - // dose state untouched, so it must not wake the user's other devices. - // v1.28 — register the landed row under its external dose-event - // UUID so a repeat of the same UUID later in THIS batch dedups - // in-memory instead of tripping the partial unique. - const landed = results[results.length - 1]; - if (entry.externalId !== undefined && landed?.id !== undefined) { - rowIdByExternalId.set(entry.externalId, landed.id); - } - const lastStatus = results[results.length - 1]?.status; - if (lastStatus === "inserted" || lastStatus === "updated") { - const scheduledForIso = effectiveScheduledFor.toISOString(); - syncSlots.set(`${entry.medicationId}|${scheduledForIso}`, { - medicationId: entry.medicationId, - scheduledFor: scheduledForIso, - }); - if (!entry.skipped) snoozeClearIds.add(entry.medicationId); - } - } catch (err: unknown) { - // P2002 = unique-constraint violation. Two shapes reach here: - // 1. an idempotencyKey collision on the unscheduled/PRN insert → - // "duplicate" (the canonical-slot path already absorbs its own - // same-slot P2002 race inside `applyCanonicalSlotWrite`); - // 2. a same-slot collision on the unscheduled insert. In both - // cases we surface the existing row id so the cursor advances. - if ( - typeof err === "object" && - err !== null && - "code" in err && - (err as { code: string }).code === "P2002" - ) { - const existing = entry.idempotencyKey - ? await prisma.medicationIntakeEvent.findUnique({ - where: { - userId_idempotencyKey: { - userId: user.id, - idempotencyKey: entry.idempotencyKey, - }, - }, - select: { id: true }, - }) - : // v1.28 — an Apple entry that raced the partial - // `(user_id, external_id)` unique (concurrent batches syncing - // the same dose event) resolves to the winning live row. - entry.externalId - ? await prisma.medicationIntakeEvent.findFirst({ - where: { - userId: user.id, - externalId: entry.externalId, - deletedAt: null, - }, - select: { id: true }, - }) - : null; - duplicates += 1; - results.push({ - index: i, - status: "duplicate", - id: existing?.id, - }); - continue; - } - const reason = - err instanceof Error ? err.message.slice(0, 120) : "create_failed"; - skipped.push({ index: i, reason }); - results.push({ index: i, status: "skipped", reason }); - } - } - - await auditLog("medications.intake.bulk.ingest", { - userId: user.id, + const outcome = await ingestBulkIntake({ + user, + scoped, + entries, + syncTrigger, ipAddress: getClientIp(request), - details: { - processed: entries.length, - inserted, - updated, - duplicates, - skipped: skipped.length, - }, + originDeviceToken: request.headers.get("x-device-id"), }); - - annotate({ - action: { name: "medications.intake.bulk.ingest" }, - meta: { - processed: entries.length, - inserted, - updated, - duplicates, - skipped: skipped.length, - extra_doses_unscheduled: extraDosesUnscheduled, - ...(unstableShapes.length > 0 - ? unstableExternalIdMeta("medication.intake.bulk", unstableShapes) - : {}), - }, - }); - - // v1.4.34 — bust per-user medications + compliance + achievement - // caches when at least one row landed so the next read reflects the - // ingested batch. v1.16.8 — hard-evict: this bulk endpoint is the iOS - // user's INTERACTIVE intake path (a dose taken / skipped on the phone - // syncs through here), so the very next read must show the user their - // own action — a marked-stale SWR cell would hand back the pre-dose - // payload. The genuinely background writers (the auto-miss cron in - // `intake-auto-skip.ts`, slot dedup in `intake-slot-dedup.ts`) keep - // the default mark-stale at their own call sites. - if (inserted > 0 || updated > 0) { - invalidateUserMedications(user.id, { evict: true }); + if (outcome.kind === "refused") { + return apiError(outcome.message, 422, { errorCode: outcome.errorCode }); } - - // Refresh the derived rows of each distinct `(medicationId, dayKey)` the - // batch touched. Best-effort: the refresh reports its own failures and - // never fails the ingest. - for (const { medicationId, dayKey } of touchedDays.values()) { - await refreshDoseDay({ - userId: user.id, - medicationId, - dayKey, - tz: user.timezone, - }); - } - - // v1.17.1 (#22) — silent cross-device intake sync. ONE queued fan-out - // for the whole batch regardless of how many slots it touched (the - // per-user coalescer folds bursts further), excluding the originating - // device (`X-Device-Id` = registered `Device.token`). APNs-only, - // best-effort, fire-and-forget: the canonical rows are already - // persisted, so a sync-push miss never affects the batch response. - // The single-intake routes reset a medication's snooze on every recorded - // take; a drained offline batch must quiet the snooze the same way, or a - // dose synced in later keeps counting as snoozed although it was taken. - if (snoozeClearIds.size > 0) { - await prisma.medication.updateMany({ - where: { id: { in: [...snoozeClearIds] }, userId: user.id }, - data: { snoozedUntil: null }, - }); - } - - if (syncSlots.size > 0) { - queueMedicationIntakeSync({ - userId: user.id, - originDeviceToken: request.headers.get("x-device-id"), - }); - - // v1.18.4 — PWA-only counterpart: close the still-pending dose-due - // reminders for every affected slot over Web Push and refresh the app - // badge once with the post-batch outstanding-dose count. Best-effort, - // fire-and-forget — a clear miss never affects the batch response. - const affectedSlots = Array.from(syncSlots.values()); - void (async () => { - const badgeCount = await countOutstandingDosesToday( - user.id, - user.timezone, - ); - await dispatchMedicationIntakeWebClearBulk({ - userId: user.id, - slots: affectedSlots, - badgeCount, - }); - })(); - } - - // v1.42.4 — Apple Health doses get the per-type arrival record the - // measurement batch keeps for its types: when they last arrived, under which - // trigger, when a sync last brought a new or changed dose, and when a - // background sync last carried them. Entries sent with `source: - // "APPLE_HEALTH"` only; a dose taken in the app is not an Apple Health - // arrival. Replays answered `duplicate` alone are not one (#1259), the - // same rule as the batch route. - const doseArrival = { accepted: 0, newSamples: 0 }; - for (const result of results) { - if (entries[result.index]?.source !== "APPLE_HEALTH") continue; - if (result.status === "skipped") continue; - doseArrival.accepted += 1; - if (result.status !== "duplicate") doseArrival.newSamples += 1; - } - if (doseArrival.accepted > 0 && !scoped) { - await recordHealthKitTypeArrivals(prisma, { - userId: user.id, - counts: new Map([[MEDICATION_DOSE_ARRIVAL_TYPE, doseArrival]]), - trigger: syncTrigger ?? null, - }); - } - - // The SyncMode checkpoint belongs to the push that moved the data, not to - // the `/api/sync/state` handshake that reports it. This endpoint is one of - // the three domains that handshake summarises, so an intake drain advances - // it. A `backfill` catch-up does not: it is history, not a sync. - if (syncTrigger !== BACKFILL_SYNC_TRIGGER) { - await markSyncCheckpoint(user.id); - } - - return apiSuccess({ - processed: entries.length, - inserted, - updated, - duplicates, - skipped, - entries: results, - }); + return apiSuccess(outcome.body); } diff --git a/src/lib/medications/intake-bulk/batch-state.ts b/src/lib/medications/intake-bulk/batch-state.ts new file mode 100644 index 000000000..d9d1f2397 --- /dev/null +++ b/src/lib/medications/intake-bulk/batch-state.ts @@ -0,0 +1,102 @@ +/** + * What a bulk intake batch accumulates while its entries are walked: the + * per-entry results and counters the response carries, and the distinct + * days, slots and medications the tails after the loop act on. + */ +import type { UnstableExternalIdShape } from "@/lib/validations/external-id"; +import type { InjectionSiteValue } from "@/lib/validations/medication"; + +/** + * One entry of `POST /api/medications/intake/bulk`, as the route's schema + * parses it. + */ +export interface BulkIntakeEntry { + medicationId: string; + scheduledFor?: Date; + takenAt?: Date; + skipped: boolean; + idempotencyKey?: string; + injectionSite?: InjectionSiteValue; + forceSlotInstant?: Date; + doseTaken?: string; + source?: "APPLE_HEALTH"; + externalId?: string; +} + +/** + * `updated` joins the per-entry status vocabulary: a write that snaps onto + * an existing scheduled-slot row (e.g. the pending REMINDER row) updates it + * in place rather than inserting. The iOS sync engine treats `updated` the + * same as `inserted` for cursor advancement — both mean "the server accepted + * this entry and produced a row id". + */ +type EntryStatus = "inserted" | "updated" | "duplicate" | "skipped"; +export interface EntryResult { + index: number; + status: EntryStatus; + reason?: string; + id?: string; +} + +export interface BatchState { + results: EntryResult[]; + inserted: number; + updated: number; + duplicates: number; + /** + * External dose records kept as their own unscheduled dose because their + * band slot already carried a different one. + */ + extraDosesUnscheduled: number; + skipped: Array<{ index: number; reason: string }>; + unstableShapes: UnstableExternalIdShape[]; + /** + * Distinct `(medicationId, dayKey)` pairs touched by the batch, so one + * dose-day refresh fires per pair after all inserts complete. A per-row + * refresh would balloon a 500-entry batch into 500 sequential refreshes. + */ + touchedDays: Map; + /** + * Distinct `(medicationId, scheduledFor)` slots whose state actually + * changed in this batch (#22: a row inserted / updated), so the silent + * cross-device sync fires once per affected slot — not once per row, and + * not for pending echoes / duplicates that changed nothing. + */ + syncSlots: Map; + /** + * A landed non-skipped dose quiets an active snooze, exactly like the + * single-intake routes (they null `snoozedUntil` on every recorded take). + * Collected per medication here, cleared once after the loop. + */ + snoozeClearIds: Set; + /** + * Live rows already carrying one of the batch's external dose-event UUIDs, + * and the rows this batch landed under one, so an intra-batch repeat of a + * UUID dedups too. + */ + rowIdByExternalId: Map; +} + +export function createBatchState( + rowIdByExternalId: Map, +): BatchState { + return { + results: [], + inserted: 0, + updated: 0, + duplicates: 0, + extraDosesUnscheduled: 0, + skipped: [], + unstableShapes: [], + touchedDays: new Map(), + syncSlots: new Map(), + snoozeClearIds: new Set(), + rowIdByExternalId, + }; +} + +/** Record an entry the batch skipped, with its reason. */ +export function skipEntry(state: BatchState, index: number, reason: string) { + state.skipped.push({ index, reason }); + state.results.push({ index, status: "skipped", reason }); +} diff --git a/src/lib/medications/intake-bulk/entry-attribution.ts b/src/lib/medications/intake-bulk/entry-attribution.ts new file mode 100644 index 000000000..aa509ba96 --- /dev/null +++ b/src/lib/medications/intake-bulk/entry-attribution.ts @@ -0,0 +1,178 @@ +/** + * Which slot one bulk intake entry binds to. + * + * Taken writes attribute through the window-band engine (the SAME minter the + * read ledger + the compliance % consume), so a bulk-synced take binds to a + * slot identically to the single-route write. The optional + * `forceSlotInstant` pins an off-window take onto a named real slot. Pending + * echoes and skips carry no take to attribute by, so they keep the canonical + * anchor snap that binds them to the projector/worker-minted slot row. + */ +import type { SessionUser } from "@/lib/auth/session-user"; +import { + findPinConflict, + resolveForcedSlotForWrite, + resolveSlotForWriteByBand, + resolveSlotInstantForWrite, + slotHeldByAnotherDose, +} from "@/lib/medications/scheduling/slot-upsert"; +import type { BulkIntakeEntry } from "./batch-state"; + +export interface EntryAttribution { + canonicalSlot: Date | null; + /** + * Binding provenance: USER_PIN on the forced path, AUTO on a band + * decision. Pending echoes / skips carry no decision (undefined → the + * upsert leaves an existing USER_PIN untouched). + */ + attributionSource: "AUTO" | "USER_PIN" | undefined; + /** + * Set when the band slot is held by a different dose and this take is + * recorded unscheduled; it must not converge onto any slot row. + */ + heldByAnotherDose: boolean; +} + +interface EntryContext { + user: Pick; + entry: BulkIntakeEntry; + incomingScheduledFor: Date; +} + +async function attributePin( + ctx: EntryContext, + forceSlotInstant: Date, +): Promise { + const { user, entry } = ctx; + const canonicalSlot = await resolveForcedSlotForWrite({ + userId: user.id, + medicationId: entry.medicationId, + userTz: user.timezone, + slotInstant: forceSlotInstant, + }); + if (canonicalSlot === null) { + // Mirrors the single route's 422 (`medications.intake.force_slot + // .invalid`) as a per-entry skip so one bad pin never fails the + // batch. + return { reason: "force_slot_invalid" }; + } + // Refuse to pin onto a slot another recorded action + // already serves: the explicit-write last-write-wins rule would + // silently overwrite that dose record. Per-entry skip, mirroring + // the single route's 422 `medications.intake.force_slot.occupied`. + if ( + await findPinConflict({ + userId: user.id, + medicationId: entry.medicationId, + canonicalSlot, + incomingTakenAt: entry.takenAt ?? null, + }) + ) { + return { reason: "force_slot_occupied" }; + } + return { + canonicalSlot, + attributionSource: "USER_PIN", + heldByAnotherDose: false, + }; +} + +async function attributeTake(ctx: EntryContext): Promise { + const { user, entry } = ctx; + const attribution = await resolveSlotForWriteByBand({ + userId: user.id, + medicationId: entry.medicationId, + userTz: user.timezone, + takenAt: entry.takenAt as Date, + }); + let canonicalSlot = attribution.slotInstant; + let heldByAnotherDose = false; + // An external dose record (a HealthKit dose event) whose band slot + // already carries a different recorded dose is an extra dose, not a + // correction of that one. Converging would overwrite the slot's + // take last-write-wins and the extra dose would exist nowhere; it is + // recorded as its own unscheduled dose instead. + if ( + canonicalSlot !== null && + entry.externalId !== undefined && + (await slotHeldByAnotherDose({ + userId: user.id, + medicationId: entry.medicationId, + canonicalSlot, + takenAt: entry.takenAt as Date, + externalId: entry.externalId, + scheduledFor: entry.scheduledFor ?? null, + userTz: user.timezone, + })) + ) { + canonicalSlot = null; + heldByAnotherDose = true; + } + return { canonicalSlot, attributionSource: "AUTO", heldByAnotherDose }; +} + +/** + * The slot of one entry, or the per-entry skip reason of a bad pin. The + * write flags say what the entry is: an offline-sync replay echoes a PENDING + * projection (no `takenAt`, `skipped:false`) for a slot the user may already + * have actioned; an explicit `takenAt` or `skipped:true` is a real user + * action and applies last-write-wins. + */ +export async function attributeEntry( + ctx: EntryContext, + flags: { isExplicitTaken: boolean; isExplicitSkip: boolean }, +): Promise { + const { user, entry, incomingScheduledFor } = ctx; + if (flags.isExplicitTaken && entry.forceSlotInstant !== undefined) { + return attributePin(ctx, entry.forceSlotInstant); + } + if (flags.isExplicitTaken) { + return attributeTake(ctx); + } + if (flags.isExplicitSkip && entry.scheduledFor === undefined) { + // A slot-less deliberate skip resolves by band + // membership on the skip moment, exactly like a take does. The + // defaulted-now snap below only reaches the tight reminder-grace + // window, so a skip tapped 20 minutes past the dose time inserted + // an orphan ad-hoc skip row while the slot's pending REMINDER row + // stayed open and later auto-missed — the user's deliberate skip + // recorded as a miss. + const attribution = await resolveSlotForWriteByBand({ + userId: user.id, + medicationId: entry.medicationId, + userTz: user.timezone, + takenAt: incomingScheduledFor, + }); + return { + canonicalSlot: attribution.slotInstant, + attributionSource: undefined, + heldByAnotherDose: false, + }; + } + // Source-agnostic anchor snap for pending echoes + skips. + // iOS posts the reminder actions here (source API); without this + // they inserted a SECOND row for a slot already carrying the + // projector/worker's pending REMINDER row (the unique key includes + // `source`, and the iOS-vs-server `scheduledFor` drifts by a + // minute). Snap onto the canonical slot instant and update the + // existing row in place. + const canonicalSlot = await resolveSlotInstantForWrite({ + userId: user.id, + medicationId: entry.medicationId, + userTz: user.timezone, + incoming: incomingScheduledFor, + // Only a client-supplied `scheduledFor` names a real slot. A + // defaulted-now write must not snap across the wide ±halfGap + // window onto a far slot (phantom morning dose). + instantIsExplicit: entry.scheduledFor !== undefined, + // Pending echoes / skips never record a take; a pending sync + // echo legitimately maps to a future slot, so the taken + // future-slot guard stays off here. + isTakenWrite: false, + }); + return { + canonicalSlot, + attributionSource: undefined, + heldByAnotherDose: false, + }; +} diff --git a/src/lib/medications/intake-bulk/entry-write.ts b/src/lib/medications/intake-bulk/entry-write.ts new file mode 100644 index 000000000..2a6c6b045 --- /dev/null +++ b/src/lib/medications/intake-bulk/entry-write.ts @@ -0,0 +1,251 @@ +/** + * The row one bulk intake entry lands on: the canonical slot write, the + * idempotency replay, the convergence probe onto a row already on the named + * instant, or the standalone insert, with the inventory hook each one owes. + */ +import { prisma } from "@/lib/db"; +import type { SessionUser } from "@/lib/auth/session-user"; +import type { InjectionSiteKey } from "@/lib/medications/injection-sites"; +import { + applyCanonicalSlotWrite, + mayConvergeOntoSuppliedSlot, +} from "@/lib/medications/scheduling/slot-upsert"; +import { + consumeForIntake, + restoreForIntake, +} from "@/lib/medications/inventory/consumption"; +import type { BatchState, BulkIntakeEntry } from "./batch-state"; +import type { EntryAttribution } from "./entry-attribution"; + +export interface EntryWrite { + user: Pick; + entry: BulkIntakeEntry; + index: number; + incomingScheduledFor: Date; + isExplicitTaken: boolean; + isExplicitSkip: boolean; + attribution: EntryAttribution; + resolvedInjectionSite: InjectionSiteKey | null; +} + +/** + * Converge onto the one slot row through the shared upsert: H1 + * deterministic selection, C2 no-downgrade guard, and a C1 race-safe create + * that re-finds + updates on a P2002 collision rather than misclassifying it + * as a duplicate and dropping the dose. + */ +async function writeOntoSlot( + write: EntryWrite, + state: BatchState, + canonicalSlot: Date, +): Promise { + const { user, entry, index: i, isExplicitTaken, isExplicitSkip } = write; + const applied = await applyCanonicalSlotWrite({ + client: prisma, + userId: user.id, + medicationId: entry.medicationId, + canonicalSlot, + takenAt: entry.takenAt ?? null, + skipped: entry.skipped, + isExplicitTaken, + isExplicitSkip, + idempotencyKey: entry.idempotencyKey ?? null, + // An Apple dose-event import stamps its provenance on a + // fresh row; every other caller keeps today's "API". + createSource: entry.source ?? "API", + injectionSite: write.resolvedInjectionSite, + attributionSource: write.attribution.attributionSource, + // Dose override only documents a consumed dose. + doseTaken: (!entry.skipped && entry.doseTaken) || null, + // External dose-event UUID rides onto the landed row. + externalId: entry.externalId ?? null, + }); + if (applied.noDowngradeNoOp) { + // Pending echo onto an already-actioned slot. Report it as a + // duplicate so the iOS cursor advances WITHOUT downgrading the + // recorded dose. A pending echo is never a taken write, so it + // must never reach the inventory consume below. + state.duplicates += 1; + state.results.push({ index: i, status: "duplicate", id: applied.row.id }); + } else if (applied.outcome === "updated") { + state.updated += 1; + state.results.push({ index: i, status: "updated", id: applied.row.id }); + } else { + state.inserted += 1; + state.results.push({ index: i, status: "inserted", id: applied.row.id }); + } + // Inventory: only the genuine pending→taken + // transition consumes (the upsert's `consumedTransition` flag — + // the ONLY transition that may move stock). The stamp keeps a + // replayed batch exactly-once for stamped rows; the + // transition gate additionally protects pre-stamp rows: a full + // journal replay re-posting historical taken doses must not + // drain today's stock through their NULL stamps. An explicit + // skip refunds a previously-taken row's stamp. Pending echoes + // carry neither flag and never touch stock. + if (applied.consumedTransition && isExplicitTaken) { + await consumeForIntake({ + client: prisma, + userId: user.id, + medicationId: entry.medicationId, + eventId: applied.row.id, + intakeAt: entry.takenAt as Date, + }); + } else if (!applied.noDowngradeNoOp && isExplicitSkip) { + await restoreForIntake({ + client: prisma, + userId: user.id, + eventId: applied.row.id, + }); + } +} + +/** + * The idempotencyKey replay pre-check: a re-submission returns the existing + * row as "duplicate" (previously reached via the P2002 catch; the explicit + * probe keeps that contract now that a replay could otherwise converge as + * "updated"). Scoped to the caller. The key is client-supplied, so an + * unscoped read would resolve another user's row and hand its id back here + * (migration 0273 made the unique per user). + */ +async function replayedRowId(write: EntryWrite): Promise { + const { user, entry } = write; + if (!entry.idempotencyKey) return null; + const replay = await prisma.medicationIntakeEvent.findUnique({ + where: { + userId_idempotencyKey: { + userId: user.id, + idempotencyKey: entry.idempotencyKey, + }, + }, + select: { id: true }, + }); + return replay?.id ?? null; +} + +/** + * Genuinely standalone. Anchor the row on the intake instant — the + * documented ad-hoc contract (`scheduledFor = takenAt`) — so an unresolvable + * client anchor (a future slot instant the taken guard rejected) can never + * park a live row exactly on a slot the worker later mints a pending + * REMINDER row for. A pending echo (no takenAt) keeps the incoming instant. + * The idempotencyKey, when supplied, has a UNIQUE index; a racing + * re-submission returns the existing row via P2002 → status: "duplicate". + */ +async function insertStandalone( + write: EntryWrite, + state: BatchState, +): Promise { + const { user, entry, index: i, resolvedInjectionSite } = write; + const effectiveScheduledFor = entry.takenAt ?? write.incomingScheduledFor; + const row = await prisma.medicationIntakeEvent.create({ + data: { + userId: user.id, + medicationId: entry.medicationId, + scheduledFor: effectiveScheduledFor, + takenAt: entry.takenAt ?? null, + skipped: entry.skipped, + // Apple provenance on a standalone (ad-hoc) row; + // every other caller keeps today's "API". + source: entry.source ?? "API", + idempotencyKey: entry.idempotencyKey ?? null, + // External dose-event UUID for the idempotent + // re-sync dedup. + ...(entry.externalId !== undefined && { + externalId: entry.externalId, + }), + // Site only on a resolved taken-injection entry. + ...(resolvedInjectionSite !== null && { + injectionSite: resolvedInjectionSite, + }), + // Dose override only on a taken entry carrying one. + ...(!entry.skipped && entry.doseTaken && { doseTaken: entry.doseTaken }), + }, + }); + state.inserted += 1; + state.results.push({ index: i, status: "inserted", id: row.id }); + // A fresh standalone taken row consumes inventory + // units (a fresh skip / pending row has nothing to consume or + // refund). + if (write.isExplicitTaken) { + await consumeForIntake({ + client: prisma, + userId: user.id, + medicationId: entry.medicationId, + eventId: row.id, + intakeAt: entry.takenAt as Date, + }); + } + return effectiveScheduledFor; +} + +/** + * Write the entry. The answer is the `scheduledFor` instant the row actually + * lands on, so the dose-day refresh keys the same day the stored row anchors + * to, or null when an idempotency replay answered the entry and nothing + * after the write applies to it. + */ +export async function writeEntry( + write: EntryWrite, + state: BatchState, +): Promise { + const { user, entry, index: i, incomingScheduledFor } = write; + const { canonicalSlot, heldByAnotherDose } = write.attribution; + if (canonicalSlot) { + // Scheduled dose — converge onto the one canonical slot row. + await writeOntoSlot(write, state, canonicalSlot); + return canonicalSlot; + } + // Resolver-null: PRN / off-slot / future-slot-guarded. Before the + // standalone insert, two guards keep the slot from forking into a + // second live row: + // + // 1. idempotencyKey replay pre-check (see `replayedRowId`); + // 2. source-agnostic convergence probe — when ANY live row already + // sits on the incoming instant (e.g. the pending REMINDER row + // the worker pre-minted on a slot the taken-write future guard + // refused to snap to), converge onto THAT row through the shared + // slot upsert instead of inserting a sibling that differs only + // by `source`. The partial unique index carries `source`, so it + // cannot catch this duplicate; the probe must. + const replayId = await replayedRowId(write); + if (replayId !== null) { + state.duplicates += 1; + state.results.push({ index: i, status: "duplicate", id: replayId }); + return null; + } + // Only an explicit client `scheduledFor` names a slot a row could + // already sit on; a defaulted anchor (takenAt / now) never does, so + // the probe is skipped on that path. + // + // Dose-safety guard: a TAKEN entry must not converge onto a slot + // whose anchor is in the future relative to the take. The resolver's + // forward guard already refused to snap a late take onto a future + // slot; the probe must honour the same invariant rather than binding + // the take forward onto that slot's pending REMINDER row (a + // late-morning dose consuming the evening slot). It records + // standalone (ad-hoc) instead. + const existingSlotRow = + !heldByAnotherDose && + entry.scheduledFor !== undefined && + mayConvergeOntoSuppliedSlot({ + skipped: entry.skipped, + takenAt: entry.takenAt ?? null, + suppliedSlot: incomingScheduledFor, + }) + ? await prisma.medicationIntakeEvent.findFirst({ + where: { + userId: user.id, + medicationId: entry.medicationId, + scheduledFor: incomingScheduledFor, + deletedAt: null, + }, + select: { id: true }, + }) + : null; + if (existingSlotRow) { + await writeOntoSlot(write, state, incomingScheduledFor); + return incomingScheduledFor; + } + return insertStandalone(write, state); +} diff --git a/src/lib/medications/intake-bulk/entry.ts b/src/lib/medications/intake-bulk/entry.ts new file mode 100644 index 000000000..c38998ceb --- /dev/null +++ b/src/lib/medications/intake-bulk/entry.ts @@ -0,0 +1,303 @@ +/** + * One entry of a bulk intake batch, start to end: the per-entry refusals, + * the slot attribution, the injection site, the write, and what the batch + * remembers of a row that landed. One bad entry never fails the batch; every + * refusal is that entry's own outcome. + */ +import type { Prisma } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import type { SessionUser } from "@/lib/auth/session-user"; +import { dayKeyForScheduledFor } from "@/lib/medications/dose-day-refresh"; +import { resolveInjectionSiteForWrite } from "@/lib/medications/injection-site-write"; +import type { InjectionSiteKey } from "@/lib/medications/injection-sites"; +import { + INTAKE_NOT_TRACKED_REASON, + isRecordOnly, +} from "@/lib/medications/intake-tracking"; +import { + classifyExternalId, + UNSTABLE_EXTERNAL_ID_REASON, +} from "@/lib/validations/external-id"; +import type { InjectionSiteValue } from "@/lib/validations/medication"; +import { + skipEntry, + type BatchState, + type BulkIntakeEntry, +} from "./batch-state"; +import { attributeEntry } from "./entry-attribution"; +import { writeEntry } from "./entry-write"; + +/** + * What the batch reads once about the medications it names: the + * injection-tracking fields, so a per-entry `injectionSite` can be resolved + + * validated without an extra per-entry read, and `externalSource`, so the + * Apple-entry mirror gate needs no extra read either. + */ +export const OWNED_MEDICATION_SELECT = { + id: true, + deliveryForm: true, + trackInjectionSites: true, + allowedInjectionSites: true, + externalSource: true, + trackIntake: true, +} satisfies Prisma.MedicationSelect; + +export type OwnedMedication = Prisma.MedicationGetPayload<{ + select: typeof OWNED_MEDICATION_SELECT; +}>; + +export interface BatchContext { + user: Pick; + ownedSet: Set; + medById: Map; + globalExcluded: InjectionSiteValue[]; +} + +/** + * The refusals decided before any slot is read. True when the entry is + * answered (skipped, or a duplicate of a row the batch already knows). + */ +function screenEntry( + ctx: BatchContext, + state: BatchState, + entry: BulkIntakeEntry, + i: number, +): boolean { + if (!ctx.ownedSet.has(entry.medicationId)) { + skipEntry(state, i, "medication_not_found"); + return true; + } + + // A record-only medication takes no new dose (iOS #116, item 15). The + // single-dose routes answer 422 `medication.intake.notTracked`; here the + // same decision is one entry's outcome, so the rest of the batch lands. + if (isRecordOnly(ctx.medById.get(entry.medicationId) ?? {})) { + skipEntry(state, i, INTAKE_NOT_TRACKED_REASON); + return true; + } + + // Both dedup keys on this route are only idempotent while the value + // is STABLE across client launches: `(userId, externalId)` for the + // Apple dose event, and the `@unique` `idempotency_key` column for + // the replay probe. A value that rotates per process (an + // object description carrying a memory address) never matches its + // own earlier row, so every re-sync logs the dose again. Refused per + // entry — one bad row must not stop the rest of the batch landing. + const unstable = + (entry.externalId === undefined + ? null + : classifyExternalId(entry.externalId)) ?? + (entry.idempotencyKey === undefined + ? null + : classifyExternalId(entry.idempotencyKey)); + if (unstable) { + state.unstableShapes.push(unstable); + skipEntry(state, i, UNSTABLE_EXTERNAL_ID_REASON); + return true; + } + + // Apple dose-event replay: the batch pre-check found a live + // row already carrying this external UUID (or an earlier entry in + // THIS batch inserted it). First-write-wins per Apple dose — report + // `duplicate` with the existing row id so the sync cursor advances. + if (entry.externalId !== undefined) { + const existingRowId = state.rowIdByExternalId.get(entry.externalId); + if (existingRowId !== undefined) { + state.duplicates += 1; + state.results.push({ index: i, status: "duplicate", id: existingRowId }); + return true; + } + } + return false; +} + +/** + * Resolve + validate the optional per-entry injection site. A disallowed + * site marks the entry skipped without failing the batch (per-entry + * contract); a non-injection / tracking-off / non-taken entry silently drops + * it. + */ +function resolveEntrySite( + ctx: BatchContext, + entry: BulkIntakeEntry, + isExplicitTaken: boolean, +): { site: InjectionSiteKey | null } | { reason: string } { + let resolvedInjectionSite: InjectionSiteKey | null = null; + if (entry.injectionSite !== undefined) { + const med = ctx.medById.get(entry.medicationId); + const resolution = resolveInjectionSiteForWrite({ + submitted: entry.injectionSite, + taken: isExplicitTaken, + deliveryForm: med?.deliveryForm ?? "ORAL", + trackInjectionSites: med?.trackInjectionSites ?? false, + allowedInjectionSites: (med?.allowedInjectionSites ?? + []) as InjectionSiteKey[], + globalExcludedInjectionSites: ctx.globalExcluded as InjectionSiteKey[], + }); + if (resolution.kind === "disallowed") { + return { reason: "injection_site_not_allowed" }; + } + resolvedInjectionSite = resolution.site; + } + return { site: resolvedInjectionSite }; +} + +/** What the batch remembers of a row that landed. */ +function recordLanded( + ctx: BatchContext, + state: BatchState, + entry: BulkIntakeEntry, + effectiveScheduledFor: Date, +): void { + const dayKey = dayKeyForScheduledFor( + effectiveScheduledFor, + ctx.user.timezone, + ); + state.touchedDays.set(`${entry.medicationId}|${dayKey}`, { + medicationId: entry.medicationId, + dayKey, + }); + // Collect the slot for the silent cross-device sync (#22) + // only when the row's state actually changed (inserted / updated). A + // pending echo, duplicate, or no-downgrade no-op leaves the visible + // dose state untouched, so it must not wake the user's other devices. + // Register the landed row under its external dose-event + // UUID so a repeat of the same UUID later in THIS batch dedups + // in-memory instead of tripping the partial unique. + const landed = state.results[state.results.length - 1]; + if (entry.externalId !== undefined && landed?.id !== undefined) { + state.rowIdByExternalId.set(entry.externalId, landed.id); + } + const lastStatus = state.results[state.results.length - 1]?.status; + if (lastStatus === "inserted" || lastStatus === "updated") { + const scheduledForIso = effectiveScheduledFor.toISOString(); + state.syncSlots.set(`${entry.medicationId}|${scheduledForIso}`, { + medicationId: entry.medicationId, + scheduledFor: scheduledForIso, + }); + if (!entry.skipped) state.snoozeClearIds.add(entry.medicationId); + } +} + +async function writeScreenedEntry( + ctx: BatchContext, + state: BatchState, + entry: BulkIntakeEntry, + i: number, +): Promise { + const incomingScheduledFor = + entry.scheduledFor ?? entry.takenAt ?? new Date(); + + // Classify the incoming write. An offline-sync replay echoes a + // PENDING projection (no `takenAt`, `skipped:false`) for a slot the + // user may already have actioned; that echo must NEVER clear a + // recorded `takenAt`. An explicit `takenAt` or `skipped:true` is a + // real user action and applies last-write-wins. + const isExplicitTaken = !entry.skipped && entry.takenAt !== undefined; + const isExplicitSkip = entry.skipped === true; + + const attribution = await attributeEntry( + { user: ctx.user, entry, incomingScheduledFor }, + { isExplicitTaken, isExplicitSkip }, + ); + if ("reason" in attribution) { + skipEntry(state, i, attribution.reason); + return; + } + if (attribution.heldByAnotherDose) state.extraDosesUnscheduled += 1; + + const site = resolveEntrySite(ctx, entry, isExplicitTaken); + if ("reason" in site) { + skipEntry(state, i, site.reason); + return; + } + + const effectiveScheduledFor = await writeEntry( + { + user: ctx.user, + entry, + index: i, + incomingScheduledFor, + isExplicitTaken, + isExplicitSkip, + attribution, + resolvedInjectionSite: site.site, + }, + state, + ); + if (effectiveScheduledFor === null) return; + recordLanded(ctx, state, entry, effectiveScheduledFor); +} + +/** + * P2002 = unique-constraint violation. Two shapes reach here: + * 1. an idempotencyKey collision on the unscheduled/PRN insert → + * "duplicate" (the canonical-slot path already absorbs its own + * same-slot P2002 race inside `applyCanonicalSlotWrite`); + * 2. a same-slot collision on the unscheduled insert. In both + * cases we surface the existing row id so the cursor advances. + */ +async function answerFailedEntry( + ctx: BatchContext, + state: BatchState, + entry: BulkIntakeEntry, + i: number, + err: unknown, +): Promise { + const { user } = ctx; + if ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code: string }).code === "P2002" + ) { + const existing = entry.idempotencyKey + ? await prisma.medicationIntakeEvent.findUnique({ + where: { + userId_idempotencyKey: { + userId: user.id, + idempotencyKey: entry.idempotencyKey, + }, + }, + select: { id: true }, + }) + : // An Apple entry that raced the partial + // `(user_id, external_id)` unique (concurrent batches syncing + // the same dose event) resolves to the winning live row. + entry.externalId + ? await prisma.medicationIntakeEvent.findFirst({ + where: { + userId: user.id, + externalId: entry.externalId, + deletedAt: null, + }, + select: { id: true }, + }) + : null; + state.duplicates += 1; + state.results.push({ + index: i, + status: "duplicate", + id: existing?.id, + }); + return; + } + const reason = + err instanceof Error ? err.message.slice(0, 120) : "create_failed"; + skipEntry(state, i, reason); +} + +/** Ingest one entry into the batch state. */ +export async function ingestEntry( + ctx: BatchContext, + state: BatchState, + entry: BulkIntakeEntry, + i: number, +): Promise { + if (screenEntry(ctx, state, entry, i)) return; + try { + await writeScreenedEntry(ctx, state, entry, i); + } catch (err: unknown) { + await answerFailedEntry(ctx, state, entry, i, err); + } +} diff --git a/src/lib/medications/intake-bulk/ingest.ts b/src/lib/medications/intake-bulk/ingest.ts new file mode 100644 index 000000000..30c2d4e27 --- /dev/null +++ b/src/lib/medications/intake-bulk/ingest.ts @@ -0,0 +1,321 @@ +/** + * Ingest one bulk intake batch: `POST /api/medications/intake/bulk`. + * + * The route authenticates, meters its own bulk bucket, reads the body and + * answers the size and schema refusals; the ownership read, the Apple mirror + * gate, the batch pre-reads, the per-entry loop and every tail after it run + * here, in the order they always ran. The outcome is the batch refusal the + * route answers with `apiError`, or the always-200 batch envelope. + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import type { SessionUser } from "@/lib/auth/session-user"; +import { markSyncCheckpoint } from "@/lib/sync/checkpoint"; +import { + BACKFILL_SYNC_TRIGGER, + MEDICATION_DOSE_ARRIVAL_TYPE, + recordHealthKitTypeArrivals, + type HealthKitRequestSyncTrigger, +} from "@/lib/integrations/healthkit-type-sync"; +import { invalidateUserMedications } from "@/lib/cache/invalidate"; +import { refreshDoseDay } from "@/lib/medications/dose-day-refresh"; +import { unstableExternalIdMeta } from "@/lib/validations/external-id"; +import type { InjectionSiteValue } from "@/lib/validations/medication"; +import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync"; +import { dispatchMedicationIntakeWebClearBulk } from "@/lib/notifications/web-push-clear"; +import { countOutstandingDosesToday } from "@/lib/medications/outstanding-doses"; +import { + createBatchState, + type BatchState, + type BulkIntakeEntry, +} from "./batch-state"; +import { + ingestEntry, + OWNED_MEDICATION_SELECT, + type BatchContext, +} from "./entry"; + +interface BatchInput { + user: Pick; + /** + * A narrow, single-purpose credential (a bridge, an ingest token) is no + * Apple Health sync, so it never writes the dose arrival record. + */ + scoped: boolean; + entries: BulkIntakeEntry[]; + syncTrigger: HealthKitRequestSyncTrigger | undefined; + ipAddress: string | null; + originDeviceToken: string | null; +} + +export type BulkIntakeOutcome = + | { kind: "refused"; message: string; errorCode: string } + | { kind: "ingested"; body: Record }; + +/** + * Idempotent re-sync dedup: one read collects every live row already + * carrying one of the batch's external dose-event UUIDs; the per-entry loop + * reports those entries as `duplicate` (first-write-wins per Apple dose) + * without touching the DB again. Fresh inserts register into the same map so + * an intra-batch repeat of a UUID dedups too. The partial + * `(user_id, external_id)` unique (migration 0239) is the race backstop. + */ +async function loadKnownExternalIds( + userId: string, + entries: BulkIntakeEntry[], +): Promise> { + const batchExternalIds = Array.from( + new Set( + entries + .map((e) => e.externalId) + .filter((id): id is string => id !== undefined), + ), + ); + const rowIdByExternalId = new Map(); + if (batchExternalIds.length > 0) { + const existingExternalRows = await prisma.medicationIntakeEvent.findMany({ + where: { + userId, + externalId: { in: batchExternalIds }, + deletedAt: null, + }, + select: { id: true, externalId: true }, + }); + for (const row of existingExternalRows) { + if (row.externalId) rowIdByExternalId.set(row.externalId, row.id); + } + } + return rowIdByExternalId; +} + +/** + * The user's global exclusion deny-list, loaded once for the whole batch (it + * is user-scoped, not per-medication). Only read when at least one entry + * actually carries an `injectionSite`, so the hot sync path (no site) pays + * for zero extra round-trips. + */ +async function loadGlobalExcluded( + userId: string, + entries: BulkIntakeEntry[], +): Promise { + const batchHasInjectionSite = entries.some( + (e) => e.injectionSite !== undefined, + ); + let globalExcluded: InjectionSiteValue[] = []; + if (batchHasInjectionSite) { + const userRow = await prisma.user.findUnique({ + where: { id: userId }, + select: { globalExcludedInjectionSites: true }, + }); + globalExcluded = (userRow?.globalExcludedInjectionSites ?? + []) as InjectionSiteValue[]; + } + return globalExcluded; +} + +/** The audit row and the event of the batch. */ +async function recordBatch(input: BatchInput, state: BatchState) { + const { entries } = input; + const { inserted, updated, duplicates, skipped, unstableShapes } = state; + await auditLog("medications.intake.bulk.ingest", { + userId: input.user.id, + ipAddress: input.ipAddress, + details: { + processed: entries.length, + inserted, + updated, + duplicates, + skipped: skipped.length, + }, + }); + + annotate({ + action: { name: "medications.intake.bulk.ingest" }, + meta: { + processed: entries.length, + inserted, + updated, + duplicates, + skipped: skipped.length, + extra_doses_unscheduled: state.extraDosesUnscheduled, + ...(unstableShapes.length > 0 + ? unstableExternalIdMeta("medication.intake.bulk", unstableShapes) + : {}), + }, + }); +} + +/** + * The caches, the dose days, the snooze, and the cross-device and web-push + * tails of the rows the batch landed. + */ +async function settleLandedRows(input: BatchInput, state: BatchState) { + const { user } = input; + // Bust per-user medications + compliance + achievement caches when at + // least one row landed so the next read reflects the ingested batch. + // Hard-evict: this bulk endpoint is the iOS user's INTERACTIVE intake path + // (a dose taken / skipped on the phone syncs through here), so the very + // next read must show the user their own action — a marked-stale SWR cell + // would hand back the pre-dose payload. The genuinely background writers + // (the auto-miss cron in `intake-auto-skip.ts`, slot dedup in + // `intake-slot-dedup.ts`) keep the default mark-stale at their own call + // sites. + if (state.inserted > 0 || state.updated > 0) { + invalidateUserMedications(user.id, { evict: true }); + } + + // Refresh the derived rows of each distinct `(medicationId, dayKey)` the + // batch touched. Best-effort: the refresh reports its own failures and + // never fails the ingest. + for (const { medicationId, dayKey } of state.touchedDays.values()) { + await refreshDoseDay({ + userId: user.id, + medicationId, + dayKey, + tz: user.timezone, + }); + } + + // Silent cross-device intake sync (#22). ONE queued fan-out + // for the whole batch regardless of how many slots it touched (the + // per-user coalescer folds bursts further), excluding the originating + // device (`X-Device-Id` = registered `Device.token`). APNs-only, + // best-effort, fire-and-forget: the canonical rows are already + // persisted, so a sync-push miss never affects the batch response. + // The single-intake routes reset a medication's snooze on every recorded + // take; a drained offline batch must quiet the snooze the same way, or a + // dose synced in later keeps counting as snoozed although it was taken. + if (state.snoozeClearIds.size > 0) { + await prisma.medication.updateMany({ + where: { id: { in: [...state.snoozeClearIds] }, userId: user.id }, + data: { snoozedUntil: null }, + }); + } + + if (state.syncSlots.size > 0) { + queueMedicationIntakeSync({ + userId: user.id, + originDeviceToken: input.originDeviceToken, + }); + + // PWA-only counterpart: close the still-pending dose-due + // reminders for every affected slot over Web Push and refresh the app + // badge once with the post-batch outstanding-dose count. Best-effort, + // fire-and-forget — a clear miss never affects the batch response. + const affectedSlots = Array.from(state.syncSlots.values()); + void (async () => { + const badgeCount = await countOutstandingDosesToday( + user.id, + user.timezone, + ); + await dispatchMedicationIntakeWebClearBulk({ + userId: user.id, + slots: affectedSlots, + badgeCount, + }); + })(); + } +} + +/** + * Apple Health doses get the per-type arrival record the measurement batch + * keeps for its types: when they last arrived, under which trigger, when a + * sync last brought a new or changed dose, and when a background sync last + * carried them. Entries sent with `source: "APPLE_HEALTH"` only; a dose taken + * in the app is not an Apple Health arrival. Replays answered `duplicate` + * alone are not one (#1259), the same rule as the batch route. Then the + * SyncMode checkpoint, which belongs to the push that moved the data, not to + * the `/api/sync/state` handshake that reports it. This endpoint is one of + * the three domains that handshake summarises, so an intake drain advances + * it. A `backfill` catch-up does not: it is history, not a sync. + */ +async function recordSyncArrival(input: BatchInput, state: BatchState) { + const { user, entries, syncTrigger } = input; + const doseArrival = { accepted: 0, newSamples: 0 }; + for (const result of state.results) { + if (entries[result.index]?.source !== "APPLE_HEALTH") continue; + if (result.status === "skipped") continue; + doseArrival.accepted += 1; + if (result.status !== "duplicate") doseArrival.newSamples += 1; + } + if (doseArrival.accepted > 0 && !input.scoped) { + await recordHealthKitTypeArrivals(prisma, { + userId: user.id, + counts: new Map([[MEDICATION_DOSE_ARRIVAL_TYPE, doseArrival]]), + trigger: syncTrigger ?? null, + }); + } + + if (syncTrigger !== BACKFILL_SYNC_TRIGGER) { + await markSyncCheckpoint(user.id); + } +} + +export async function ingestBulkIntake( + input: BatchInput, +): Promise { + const { user, entries } = input; + + // Up-front ownership check — the user can only POST intake events + // against medications they own. One read covers every medicationId + // in the batch. + const medicationIds = Array.from(new Set(entries.map((e) => e.medicationId))); + const ownedMedications = await prisma.medication.findMany({ + where: { id: { in: medicationIds }, userId: user.id }, + select: OWNED_MEDICATION_SELECT, + }); + const ownedSet = new Set(ownedMedications.map((m) => m.id)); + const medById = new Map(ownedMedications.map((m) => [m.id, m])); + + // Source-exclusive Apple-mirrored medications: an APPLE_HEALTH + // dose event may only attach to a medication mirrored from Apple Health + // (`Medication.externalSource === "APPLE_HEALTH"`). Targeting a native + // medication is a client contract violation (the iOS sync maps HK dose + // events onto the meds it mirrored itself), not a per-entry data + // problem, so the whole batch 422s with a stable errorCode the client + // can branch on. Unowned medicationIds keep the per-entry + // `medication_not_found` skip. + const appleMismatch = entries.some( + (e) => + e.source === "APPLE_HEALTH" && + ownedSet.has(e.medicationId) && + medById.get(e.medicationId)?.externalSource !== "APPLE_HEALTH", + ); + if (appleMismatch) { + return { + kind: "refused", + message: + "An APPLE_HEALTH entry may only target a medication mirrored from Apple Health", + errorCode: "medications.intake.bulk.apple_health_not_mirrored", + }; + } + + const state = createBatchState(await loadKnownExternalIds(user.id, entries)); + const ctx: BatchContext = { + user, + ownedSet, + medById, + globalExcluded: await loadGlobalExcluded(user.id, entries), + }; + + for (let i = 0; i < entries.length; i++) { + await ingestEntry(ctx, state, entries[i], i); + } + + await recordBatch(input, state); + await settleLandedRows(input, state); + await recordSyncArrival(input, state); + + return { + kind: "ingested", + body: { + processed: entries.length, + inserted: state.inserted, + updated: state.updated, + duplicates: state.duplicates, + skipped: state.skipped, + entries: state.results, + }, + }; +}