tom/HealthLog
1
0 Fork 0
mirror of https://github.com/MBombeck/HealthLog.git synced 2026-11-09 21:10:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(medications): move the bulk dose ingest out of its route

Browse Source 
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.
This commit is contained in:
Marc-André Bombeck committed 2026-10-23 14:47:57 +02:00
1 parent 2a7833d7dc
commit 24af254249
9 files changed
+1179 -833
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/__tests__/dense-measurement-read-guard.test.ts
+1
View File
@@ -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. */
src/__tests__/external-id-stability-guard.test.ts
+5 -2
View File
@@ -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",
src/__tests__/medication-intake-tracking-guard.test.ts
+1 -1
View File
@@ -72,7 +72,7 @@ const INTAKE_MAY_SEE_EVERY_ROW: Record<string, string> = {
"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":
src/app/api/medications/intake/bulk/route.ts
+17 -830
View File
@@ -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<Response> {
@@ -203,8 +150,8 @@ async function postBulk(request: NextRequest): Promise<Response> {
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<Response> {
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<Response> {
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<Response> {
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<string, string>();
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<string>();
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);
}
src/lib/medications/intake-bulk/batch-state.ts
+102
View File
@@ -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<string, { medicationId: string; dayKey: string }>;
/**
* 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<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.
*/
snoozeClearIds: Set<string>;
/**
* 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<string, string>;
}
export function createBatchState(
rowIdByExternalId: Map<string, string>,
): BatchState {
return {
results: [],
inserted: 0,
updated: 0,
duplicates: 0,
extraDosesUnscheduled: 0,
skipped: [],
unstableShapes: [],
touchedDays: new Map(),
syncSlots: new Map(),
snoozeClearIds: new Set<string>(),
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 });
}
src/lib/medications/intake-bulk/entry-attribution.ts
+178
View File
@@ -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<SessionUser, "id" | "timezone">;
entry: BulkIntakeEntry;
incomingScheduledFor: Date;
}
async function attributePin(
ctx: EntryContext,
forceSlotInstant: Date,
): Promise<EntryAttribution | { reason: string }> {
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<EntryAttribution> {
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<EntryAttribution | { reason: string }> {
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,
};
}
src/lib/medications/intake-bulk/entry-write.ts
+251
View File
@@ -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<SessionUser, "id" | "timezone">;
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<void> {
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<string | null> {
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<Date> {
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<Date | null> {
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);
}
src/lib/medications/intake-bulk/entry.ts
+303
View File
@@ -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<SessionUser, "id" | "timezone">;
ownedSet: Set<string>;
medById: Map<string, OwnedMedication>;
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<void> {
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<void> {
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<void> {
if (screenEntry(ctx, state, entry, i)) return;
try {
await writeScreenedEntry(ctx, state, entry, i);
} catch (err: unknown) {
await answerFailedEntry(ctx, state, entry, i, err);
}
}
src/lib/medications/intake-bulk/ingest.ts
+321
View File
@@ -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<SessionUser, "id" | "timezone">;
/**
* 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<string, unknown> };
/**
* 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<Map<string, string>> {
const batchExternalIds = Array.from(
new Set(
entries
.map((e) => e.externalId)
.filter((id): id is string => id !== undefined),
),
);
const rowIdByExternalId = new Map<string, string>();
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<InjectionSiteValue[]> {
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<BulkIntakeOutcome> {
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,
},
};
}