diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index 42db2e1b7..c2fc2484e 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,13 +1,12 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", - "count": 392, - "total": 139138, + "count": 388, + "total": 137367, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, "app/api/admin/notifications/reminder-check/route.ts :: apiHandler(arg)": 208, "app/api/admin/settings/route.ts :: apiHandler(arg) #2": 211, - "app/api/analytics/route.ts :: buildAnalyticsResponse": 446, "app/api/auth/login/route.ts :: apiHandler(arg)": 262, "app/api/auth/me/route.ts :: apiHandler(arg)": 386, "app/api/auth/oidc/callback/route.ts :: apiHandler(arg)": 499, @@ -37,7 +36,6 @@ "app/api/mcp/oauth/token/route.ts :: POST": 250, "app/api/mcp/oauth/token/route.ts :: POST > withBackgroundEvent(arg)": 248, "app/api/measurement-reminders/[id]/route.ts :: apiHandler(arg) #2": 239, - "app/api/measurements/series/route.ts :: apiHandler(arg)": 549, "app/api/medications/[id]/dose-history/route.ts :: apiHandler(arg)": 256, "app/api/medications/[id]/intake/route.ts :: postIntake": 607, "app/api/medications/[id]/inventory/[itemId]/route.ts :: apiHandler(arg)": 204, @@ -50,7 +48,6 @@ "app/api/medications/intake/route.ts :: apiHandler(arg) #2": 432, "app/api/mental-health/assessments/route.ts :: postAssessment": 249, "app/api/mood-entries/[id]/route.ts :: apiHandler(arg) #2": 296, - "app/api/mood-entries/bulk/route.ts :: postBulk": 479, "app/api/mood-entries/route.ts :: postMoodEntry": 324, "app/api/nutrients/batch/route.ts :: postBatch": 304, "app/api/onboarding/complete/route.ts :: completeNeedsFlow": 218, @@ -58,7 +55,6 @@ "app/api/sync/changes/route.ts :: apiHandler(arg)": 427, "app/api/user/ai-provider/route.ts :: apiHandler(arg) #2": 248, "app/api/withings/callback/route.ts :: apiHandler(arg)": 247, - "app/api/workouts/[id]/route.ts :: apiHandler(arg)": 297, "app/auth/login/page.tsx :: LoginPage": 576, "app/auth/register/page.tsx :: RegisterPage": 285, "app/coach/conversations/page.tsx :: CoachConversationsBody": 249, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index e6e413f0f..00a4bb08d 100644 --- a/src/__tests__/baselines/version-stamp-comments.json +++ b/src/__tests__/baselines/version-stamp-comments.json @@ -1,7 +1,7 @@ { "$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.", "count": 2950, - "total": 10244, + "total": 10187, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -128,7 +128,7 @@ "app/api/allergies/route.ts": 1, "app/api/analytics/__tests__/route.test.ts": 27, "app/api/analytics/range/route.ts": 2, - "app/api/analytics/route.ts": 39, + "app/api/analytics/route.ts": 10, "app/api/anamnesis/facts/__tests__/route.test.ts": 1, "app/api/auth/claim/__tests__/route.test.ts": 1, "app/api/auth/claim/preview/route.ts": 1, @@ -413,7 +413,7 @@ "app/api/measurements/series-batch/__tests__/route.test.ts": 1, "app/api/measurements/series-batch/route.ts": 1, "app/api/measurements/series/__tests__/route.test.ts": 5, - "app/api/measurements/series/route.ts": 21, + "app/api/measurements/series/route.ts": 6, "app/api/medications/__tests__/route.test.ts": 10, "app/api/medications/[id]/__tests__/route.test.ts": 8, "app/api/medications/[id]/api-endpoint/route.ts": 2, @@ -467,7 +467,7 @@ "app/api/mood-entries/bulk-delete/__tests__/route.test.ts": 1, "app/api/mood-entries/bulk-delete/route.ts": 4, "app/api/mood-entries/bulk/__tests__/route.test.ts": 5, - "app/api/mood-entries/bulk/route.ts": 16, + "app/api/mood-entries/bulk/route.ts": 6, "app/api/mood-entries/restore/__tests__/route.test.ts": 1, "app/api/mood-entries/restore/route.ts": 3, "app/api/mood-entries/route.ts": 22, @@ -559,7 +559,7 @@ "app/api/withings/webhook/route.ts": 2, "app/api/workouts/__tests__/batch-create.test.ts": 3, "app/api/workouts/__tests__/canonical-dedup.test.ts": 3, - "app/api/workouts/[id]/route.ts": 5, + "app/api/workouts/[id]/route.ts": 2, "app/api/workouts/batch/route.ts": 3, "app/api/workouts/route.ts": 2, "app/auth/claim/page.tsx": 1, diff --git a/src/__tests__/external-id-stability-guard.test.ts b/src/__tests__/external-id-stability-guard.test.ts index 75b8eb974..8b3362015 100644 --- a/src/__tests__/external-id-stability-guard.test.ts +++ b/src/__tests__/external-id-stability-guard.test.ts @@ -402,7 +402,7 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => { "lib/measurements/batch-ingest/prepare.ts", "lib/measurements/batch-ingest/ingest.ts", ], - "app/api/mood-entries/bulk/route.ts": [], + "app/api/mood-entries/bulk/route.ts": ["lib/mood/bulk-ingest.ts"], "app/api/medications/intake/bulk/route.ts": [], "app/api/workouts/batch/route.ts": [ "lib/workouts/batch-ingest/entries.ts", diff --git a/src/app/api/mood-entries/bulk/route.ts b/src/app/api/mood-entries/bulk/route.ts index b0564355e..407b3015e 100644 --- a/src/app/api/mood-entries/bulk/route.ts +++ b/src/app/api/mood-entries/bulk/route.ts @@ -20,6 +20,9 @@ * Locked contract — see `.planning/v15-ios-handoff/06-ios-responsibilities.md` * §"Cumulative metrics" sibling section for the SyncMode rationale, * and `08-locked-contracts.md` §2 for the batch envelope shape. + * + * The route authenticates, meters, caps and parses; the per-entry writes live + * in `src/lib/mood/bulk-ingest.ts`. */ import { NextRequest } from "next/server"; import { z } from "zod/v4"; @@ -27,7 +30,6 @@ import { z } from "zod/v4"; import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; -import { auditLog } from "@/lib/auth/audit"; import { apiError, apiSuccess, @@ -38,31 +40,9 @@ import { } from "@/lib/api-response"; import { withIdempotency } from "@/lib/idempotency"; import { checkRateLimit } from "@/lib/rate-limit"; -import { markSyncCheckpoint } from "@/lib/sync/checkpoint"; -import { - getScoreForMood, - moodContextSchema, - moodLevelEnum, - moodSourceEnum, -} from "@/lib/validations/mood"; -import { - classifyExternalId, - unstableExternalIdMeta, - UNSTABLE_EXTERNAL_ID_REASON, - type UnstableExternalIdShape, -} from "@/lib/validations/external-id"; -import { moodDateKey } from "@/lib/mood/date-key"; +import { moodLevelEnum, moodSourceEnum } from "@/lib/validations/mood"; import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; -import { levelAForWrite } from "@/lib/mood/level-a"; -import { persistMoodContext } from "@/lib/mood/context"; -import { encryptNote } from "@/lib/crypto/note-cipher"; -import { invalidateUserMood } from "@/lib/cache/invalidate"; -import { recomputeMoodBucketsForEntry } from "@/lib/rollups/mood-rollups"; -import { - createTagLinks, - droppedLinkKeysForWire, - NO_DROPPED_LINK_KEYS, -} from "@/lib/mood/tag-links"; +import { ingestMoodBulk } from "@/lib/mood/bulk-ingest"; import { isoDateTime } from "@/lib/validations/datetime"; const MAX_ENTRIES_PER_BATCH = 500; @@ -110,12 +90,12 @@ const bulkEntrySchema = z.object({ * to stay correct. * * `unknown` here and parsed with {@link bulkDimensionsSchema} PER ENTRY in - * the loop below, for the same reason `context` is: the doc comment beside + * the bulk ingest, for the same reason `context` is: the doc comment beside * this field used to promise that an out-of-range value marks THIS entry * skipped while the field itself sat in the batch schema and 422'd all five * hundred. The words were right and the code was not. A phone draining a * year of local history must not lose the year to one row. - * @per-entry-parse: bulkDimensionsSchema, in the loop below + * @per-entry-parse: bulkDimensionsSchema, in src/lib/mood/bulk-ingest.ts */ a1: z.unknown().optional(), a2: z.unknown().optional(), @@ -128,12 +108,12 @@ const bulkEntrySchema = z.object({ * context surface must not blank one filled in on the web. * * Declared as `unknown` here and parsed with `moodContextSchema` PER ENTRY - * in the loop below, deliberately. Validating it in the batch schema would + * in the bulk ingest, deliberately. Validating it in the batch schema would * make one entry carrying a retired key fail all five hundred, and this * endpoint's whole contract — stated in its own OpenAPI description — is * that a bad row is `skipped` and the rest of the batch lands. A phone * draining a year of local history must not lose the year to one row. - * @per-entry-parse: moodContextSchema, in the loop below + * @per-entry-parse: moodContextSchema, in src/lib/mood/bulk-ingest.ts */ context: z.unknown().optional(), moodLoggedAt: isoDateTime().transform((s) => new Date(s)), @@ -146,9 +126,9 @@ const bulkEntrySchema = z.object({ * existing `(userId, date, moodLoggedAt)` unique index still * protects against straight-up duplicates. * - * Stability is enforced per entry in the loop below, not at the field: + * Stability is enforced per entry in the bulk ingest, 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/mood-entries/bulk/route.ts + * @external-id-checked-per-entry: src/lib/mood/bulk-ingest.ts */ externalId: z.string().min(1).max(120).optional(), }); @@ -174,23 +154,6 @@ const bulkPayloadSchema = z.object({ entries: z.array(bulkEntrySchema).min(1).max(MAX_ENTRIES_PER_BATCH), }); -type EntryStatus = "inserted" | "duplicate" | "skipped"; -interface EntryResult { - index: number; - status: EntryStatus; - reason?: string; - id?: string; - // v1.12.1 — echo the client-supplied source-stable id back on each - // result so iOS can map a server row id onto its local SwiftData row - // without re-deriving it. Omitted when the entry sent no externalId. - externalId?: string; - // Submitted keys the catalog did not resolve (unknown, archived, another - // account's custom tag, or the wrong kind). Present only when non-empty, - // on an "inserted" or "duplicate" result: the entry landed, these did not. - droppedTagKeys?: string[]; - droppedFactorKeys?: string[]; -} - export const POST = apiHandler(withIdempotency<[NextRequest]>(postBulk)); async function postBulk(request: NextRequest): Promise { @@ -259,416 +222,14 @@ async function postBulk(request: NextRequest): Promise { }); } - const { entries } = parsed.data; - const tz = user.timezone ?? DEFAULT_TIMEZONE; - - const results: EntryResult[] = []; - let inserted = 0; - let duplicates = 0; - let droppedTagKeyCount = 0; - let droppedFactorKeyCount = 0; - const skipped: Array<{ index: number; reason: string }> = []; - const unstableShapes: UnstableExternalIdShape[] = []; - - // Perf — a single indexed existence read replaces the per-entry - // `findUnique` probe (up to 500 sequential round trips pre-fix). Every - // entry still resolves through exactly the key its own dedup contract - // uses (externalId when carried, else the legacy date+moodLoggedAt key), - // so lookups below are byte-identical to the per-entry probe — only the - // read is now batched. The per-entry `upsert` + tombstone check + tag - // links + reverse push all stay unchanged; batching those further would - // risk the correctness-sensitive per-row side effects for a write path - // that is usually small (a normal sync posts the current + previous - // local day, not a full historical backfill). - interface ExistingMoodRow { - id: string; - deletedAt: Date | null; - } - const orClauses: Array< - | { source: string; externalId: string } - | { date: string; moodLoggedAt: Date } - > = entries.map((entry) => - entry.externalId - ? { source: entry.source, externalId: entry.externalId } - : { - date: moodDateKey(entry.moodLoggedAt, tz), - moodLoggedAt: entry.moodLoggedAt, - }, + return apiSuccess( + await ingestMoodBulk({ + userId: user.id, + tz: user.timezone ?? DEFAULT_TIMEZONE, + entries: parsed.data.entries, + ipAddress: getClientIp(request), + parseDimensions: (dimensions) => + bulkDimensionsSchema.safeParse(dimensions), + }), ); - const existingRows = - orClauses.length > 0 - ? await prisma.moodEntry.findMany({ - where: { userId: user.id, OR: orClauses }, - select: { - id: true, - deletedAt: true, - source: true, - externalId: true, - date: true, - moodLoggedAt: true, - }, - }) - : []; - const existingByExternalKey = new Map(); - const existingByDateKey = new Map(); - for (const row of existingRows) { - if (row.externalId !== null) { - existingByExternalKey.set(`${row.source}::${row.externalId}`, row); - } - existingByDateKey.set( - `${row.date}::${row.moodLoggedAt.toISOString()}`, - row, - ); - } - - for (let i = 0; i < entries.length; i++) { - const entry = entries[i]; - - // The `(userId, source, externalId)` dedup key only holds while the - // id is STABLE across client launches; a per-process value (an object - // description carrying a memory address) mints a fresh entry on every - // re-post. Per-entry refusal, never a whole-batch 422. - const unstable = - entry.externalId === undefined - ? null - : classifyExternalId(entry.externalId); - 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, - externalId: entry.externalId, - }); - continue; - } - - // The context, parsed for THIS entry. A value outside the vocabulary - // skips this row and nothing else. - const parsedContext = - entry.context === undefined - ? { success: true as const, data: undefined } - : moodContextSchema.nullable().safeParse(entry.context); - if (!parsedContext.success) { - const reason = "invalid_context"; - skipped.push({ index: i, reason }); - results.push({ - index: i, - status: "skipped", - reason, - ...(entry.externalId ? { externalId: entry.externalId } : {}), - }); - continue; - } - - // The five dimensions, parsed for THIS entry. Out of range skips this row - // and nothing else. - const parsedDimensions = bulkDimensionsSchema.safeParse({ - a1: entry.a1, - a2: entry.a2, - a3: entry.a3, - a4: entry.a4, - a5: entry.a5, - }); - if (!parsedDimensions.success) { - const reason = "invalid_dimensions"; - skipped.push({ index: i, reason }); - results.push({ - index: i, - status: "skipped", - reason, - ...(entry.externalId ? { externalId: entry.externalId } : {}), - }); - continue; - } - - const date = moodDateKey(entry.moodLoggedAt, tz); - const score = getScoreForMood(entry.mood); - // v1.37 / v1.38 — the same resolution the single-entry path uses: - // pleasantness derived from the label unless the entry states its own, and - // the other four present only when this entry carried them. An omitted - // dimension is left alone rather than written as null, so a row that - // already carries hand-set values keeps them across a re-import. - const levelA = levelAForWrite(entry.mood, parsedDimensions.data); - - try { - // v1.12.1 — when the entry carries a source-stable `externalId`, - // dedup on the NULL-distinct `(userId, source, externalId)` key so a - // re-post with the same id (an iOS retry after a network hiccup, or a - // second adopt-on-pair backfill) updates the existing row in place - // instead of minting a duplicate when `moodLoggedAt` re-rounds / - // re-zones. Absent → the legacy `(userId, date, moodLoggedAt)` key. - // `source` carries a schema default of "MANUAL"; resolve it once so - // the probe, upsert key, and create write all agree on the value. - const resolvedSource = entry.source; - const probeWhere = entry.externalId - ? { - userId_source_externalId: { - userId: user.id, - source: resolvedSource, - externalId: entry.externalId, - }, - } - : { - userId_date_moodLoggedAt: { - userId: user.id, - date, - moodLoggedAt: entry.moodLoggedAt, - }, - }; - - // The response reliably distinguishes "inserted" from "duplicate" - // via the batched pre-read above (keyed exactly like `probeWhere`), - // so the per-entry probe here is now a Map lookup, not a round trip. - const existing = - (entry.externalId - ? existingByExternalKey.get(`${resolvedSource}::${entry.externalId}`) - : existingByDateKey.get( - `${date}::${entry.moodLoggedAt.toISOString()}`, - )) ?? null; - - // Tombstone suppression: a soft-deleted match stays deleted. The - // user-facing DELETE route flips `deletedAt` so the `/api/sync/changes` - // feed surfaces the deletion to paired clients offline at delete time - // (see the MoodEntry schema note) — an offline client's later re-post - // of the same entry is stale state, not a recreation, and must not - // resurrect the row. True no-op: no value churn, no `updatedAt` bump, - // no tag-link writes on the hidden row; report `duplicate` so the - // client checkpoints past it exactly like a live-row match. - if (existing?.deletedAt) { - duplicates += 1; - results.push({ - index: i, - status: "duplicate", - id: existing.id, - ...(entry.externalId ? { externalId: entry.externalId } : {}), - }); - continue; - } - - // The upsert and the tag/factor links run in ONE transaction, matching - // the single-entry POST path. Split apart, an out-of-scale - // `ratedFactors` value (validated inside `createTagLinks`, i.e. AFTER - // the row committed) left a persisted mood entry with no links while the - // per-entry catch reported the entry as "skipped" — the client believes - // nothing saved and a half-written row is on disk. - const written = await prisma.$transaction(async (tx) => { - const upserted = await tx.moodEntry.upsert({ - where: probeWhere, - create: { - userId: user.id, - date, - tz, - mood: entry.mood, - score, - ...levelA, - tags: entry.tags ? JSON.stringify(entry.tags) : null, - note: null, - noteEncrypted: encryptNote(entry.note ?? null), - source: resolvedSource, - externalId: entry.externalId ?? null, - moodLoggedAt: entry.moodLoggedAt, - }, - update: { - // Last-writer-wins on the mood + tags + note triple. The - // iOS client only re-posts an existing entry when it has - // new data; the server trusts that decision. When the dedup - // key is `externalId`, also refresh `date` / `moodLoggedAt` - // so a re-zoned re-import lands the corrected wall-clock on - // the same row. - mood: entry.mood, - score, - ...levelA, - tags: entry.tags ? JSON.stringify(entry.tags) : null, - note: null, - noteEncrypted: encryptNote(entry.note ?? null), - ...(entry.externalId - ? { date, moodLoggedAt: entry.moodLoggedAt } - : {}), - }, - }); - - // v1.12.0 — persist structured-tag links, mirroring the - // single-entry `createTagLinks` path. Additive + idempotent: - // `createTagLinks` resolves keys against the catalog (dropping - // unknown keys) and `skipDuplicates` on the join insert keeps a - // re-posted entry from minting duplicate links. Runs for both - // fresh and re-posted (upserted) rows so a backfill that adds tag - // keys on a second pass still lands them. - // v1.12.0 — rated factors ride the same path; an out-of-scale - // rating throws `RatedFactorOutOfRangeError`. Inside the - // transaction that throw now rolls the mood row back with it, so a - // "skipped" result means nothing was written — which is what the - // client was already being told. - // v1.38 — the day context, inside the same transaction as the row, so - // an entry reported as skipped really did write nothing. - await persistMoodContext(tx, upserted.id, user.id, parsedContext.data); - - // The keys the catalog did not resolve are reported on this entry's - // result: "inserted" alone told a client its tags had landed. - let droppedKeys = NO_DROPPED_LINK_KEYS; - if ( - (entry.tagKeys && entry.tagKeys.length > 0) || - (entry.ratedFactors && entry.ratedFactors.length > 0) - ) { - droppedKeys = await createTagLinks( - upserted.id, - user.id, - entry.tagKeys ?? [], - tx, - entry.ratedFactors ?? [], - ); - } - - return { upserted, droppedKeys }; - }); - const result = written.upserted; - const droppedKeys = written.droppedKeys; - droppedTagKeyCount += droppedKeys.droppedTagKeys.length; - droppedFactorKeyCount += droppedKeys.droppedFactorKeys.length; - - if (existing) { - duplicates += 1; - results.push({ - index: i, - status: "duplicate", - id: result.id, - ...(entry.externalId ? { externalId: entry.externalId } : {}), - ...droppedLinkKeysForWire(droppedKeys), - }); - } else { - inserted += 1; - results.push({ - index: i, - status: "inserted", - id: result.id, - ...(entry.externalId ? { externalId: entry.externalId } : {}), - ...droppedLinkKeysForWire(droppedKeys), - }); - } - } catch (err: unknown) { - // v1.37.19 — P2002 = a concurrent batch raced the same - // NULL-distinct `(userId, source, externalId)` key and won. That is - // the plain duplicate the sequential probe would have classified; - // reporting it as "skipped" with a raw Prisma message told the - // client its row was dropped when it is in fact stored. Resolve the - // winning row's id so the caller's cursor advances, exactly like - // the intake bulk route. - if ( - typeof err === "object" && - err !== null && - "code" in err && - (err as { code: string }).code === "P2002" && - entry.externalId - ) { - const winner = await prisma.moodEntry.findFirst({ - where: { - userId: user.id, - // Same resolution as the upsert key: the schema defaults - // `source` to "MANUAL" at parse time. - source: entry.source, - externalId: entry.externalId, - }, - select: { id: true }, - }); - duplicates += 1; - results.push({ - index: i, - status: "duplicate", - ...(winner ? { id: winner.id } : {}), - externalId: entry.externalId, - }); - continue; - } - const reason = - err instanceof Error ? err.message.slice(0, 120) : "upsert_failed"; - skipped.push({ index: i, reason }); - results.push({ - index: i, - status: "skipped", - reason, - ...(entry.externalId ? { externalId: entry.externalId } : {}), - }); - } - } - - await auditLog("mood.bulk.ingest", { - userId: user.id, - ipAddress: getClientIp(request), - details: { - processed: entries.length, - inserted, - duplicates, - skipped: skipped.length, - }, - }); - - annotate({ - action: { name: "mood.bulk.ingest" }, - meta: { - processed: entries.length, - inserted, - duplicates, - skipped: skipped.length, - dropped_tag_keys: droppedTagKeyCount, - dropped_factor_keys: droppedFactorKeyCount, - ...(unstableShapes.length > 0 - ? unstableExternalIdMeta("mood.bulk", unstableShapes) - : {}), - }, - }); - - // v1.4.34 — bust per-user mood + achievements + analytics caches - // when at least one row landed so the next read picks up the ingested - // batch. Skipped / duplicate-only ingests are no-ops. - if (inserted > 0) { - invalidateUserMood(user.id); - } - - // v1.4.39 — refresh the rollup tier for every distinct day - // touched by this batch. The bulk endpoint is an iOS one-shot - // backfill so the batch can span many days; we collapse to the - // unique `(user, dayStart)` set first to bound the recompute count. - // Best-effort: rollup failures must not surface as 5xx. - if (inserted > 0 || duplicates > 0) { - // v1.32.12 — collapse to the unique set of `date` labels touched by - // this batch (the same per-row `moodDateKey` the insert used), so - // the rollup keys byte-identically to the stored `MoodEntry.date`. - const touchedLabels = new Set(); - for (let i = 0; i < entries.length; i++) { - const status = results[i]?.status; - if (status === "inserted" || status === "duplicate") { - touchedLabels.add(moodDateKey(entries[i].moodLoggedAt, tz)); - } - } - try { - await Promise.all( - Array.from(touchedLabels).map((label) => - recomputeMoodBucketsForEntry(user.id, label), - ), - ); - } catch (rollupErr) { - annotate({ - meta: { - mood_rollup_write_failed: true, - mood_rollup_write_error: - rollupErr instanceof Error ? rollupErr.message : String(rollupErr), - }, - }); - } - } - - // 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 a mood drain advances it. - await markSyncCheckpoint(user.id); - - return apiSuccess({ - processed: entries.length, - inserted, - duplicates, - skipped, - entries: results, - }); } diff --git a/src/lib/mood/bulk-ingest.ts b/src/lib/mood/bulk-ingest.ts new file mode 100644 index 000000000..16b3bf76a --- /dev/null +++ b/src/lib/mood/bulk-ingest.ts @@ -0,0 +1,571 @@ +/** + * The bulk mood ingest behind `POST /api/mood-entries/bulk`. + * + * The route authenticates, meters, caps and parses the batch; this module + * writes it. Per-entry UPSERT keyed by `externalId` when the entry carries + * one, else by the legacy `(userId, date, moodLoggedAt)` key, so a re-run is + * idempotent. A bad entry is `skipped` and the rest of the batch lands: an + * unstable external id, a context outside the vocabulary, dimensions out of + * range, or a rated factor outside its own scale each refuse that one row. + */ +import type { z } from "zod/v4"; + +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import { markSyncCheckpoint } from "@/lib/sync/checkpoint"; +import { + getScoreForMood, + moodContextSchema, + type moodLevelEnum, + type moodSourceEnum, +} from "@/lib/validations/mood"; +import { + classifyExternalId, + unstableExternalIdMeta, + UNSTABLE_EXTERNAL_ID_REASON, + type UnstableExternalIdShape, +} from "@/lib/validations/external-id"; +import { moodDateKey } from "@/lib/mood/date-key"; +import { levelAForWrite, type LevelAInput } from "@/lib/mood/level-a"; +import { persistMoodContext } from "@/lib/mood/context"; +import { encryptNote } from "@/lib/crypto/note-cipher"; +import { invalidateUserMood } from "@/lib/cache/invalidate"; +import { recomputeMoodBucketsForEntry } from "@/lib/rollups/mood-rollups"; +import { + createTagLinks, + droppedLinkKeysForWire, + NO_DROPPED_LINK_KEYS, + type DroppedLinkKeys, + type RatedFactorInput, +} from "@/lib/mood/tag-links"; + +/** One entry as the route's batch schema parsed it. */ +export interface BulkMoodEntry { + mood: z.infer; + tags?: string[]; + tagKeys?: string[]; + ratedFactors?: RatedFactorInput[]; + note?: string; + a1?: unknown; + a2?: unknown; + a3?: unknown; + a4?: unknown; + a5?: unknown; + context?: unknown; + moodLoggedAt: Date; + source: z.infer; + externalId?: string; +} + +/** + * The route's per-entry dimension parse. It stays beside the batch schema + * so the bound is pinned against the single-entry one + * (`bulk-dimension-bounds.test.ts`). + */ +export type ParseBulkDimensions = ( + dimensions: Pick, +) => { success: true; data: LevelAInput } | { success: false }; + +type EntryStatus = "inserted" | "duplicate" | "skipped"; +interface EntryResult { + index: number; + status: EntryStatus; + reason?: string; + id?: string; + // Echo the client-supplied source-stable id back on each result so iOS can + // map a server row id onto its local SwiftData row without re-deriving it. + // Omitted when the entry sent no externalId. + externalId?: string; + // Submitted keys the catalog did not resolve (unknown, archived, another + // account's custom tag, or the wrong kind). Present only when non-empty, + // on an "inserted" or "duplicate" result: the entry landed, these did not. + droppedTagKeys?: string[]; + droppedFactorKeys?: string[]; +} + +export interface BulkMoodIngestResult { + processed: number; + inserted: number; + duplicates: number; + skipped: Array<{ index: number; reason: string }>; + entries: EntryResult[]; +} + +interface ExistingMoodRow { + id: string; + deletedAt: Date | null; +} + +interface ExistingRows { + byExternalKey: Map; + byDateKey: Map; +} + +/** The running totals of one batch, filled entry by entry. */ +interface Tally { + results: EntryResult[]; + skipped: Array<{ index: number; reason: string }>; + inserted: number; + duplicates: number; + droppedTagKeyCount: number; + droppedFactorKeyCount: number; + unstableShapes: UnstableExternalIdShape[]; +} + +function echoExternalId(entry: BulkMoodEntry): { externalId?: string } { + return entry.externalId ? { externalId: entry.externalId } : {}; +} + +function skip( + tally: Tally, + index: number, + entry: BulkMoodEntry, + reason: string, +): void { + tally.skipped.push({ index, reason }); + tally.results.push({ + index, + status: "skipped", + reason, + ...echoExternalId(entry), + }); +} + +/** + * Perf — a single indexed existence read replaces the per-entry `findUnique` + * probe (up to 500 sequential round trips before). Every entry still resolves + * through exactly the key its own dedup contract uses (externalId when + * carried, else the legacy date+moodLoggedAt key), so the lookups are + * byte-identical to the per-entry probe; only the read is batched. The + * per-entry `upsert` + tombstone check + tag links stay per entry; batching + * those further would risk the correctness-sensitive per-row side effects + * for a write path that is usually small (a normal sync posts the current + + * previous local day, not a full historical backfill). + */ +async function readExistingRows( + userId: string, + tz: string, + entries: BulkMoodEntry[], +): Promise { + const orClauses: Array< + | { source: string; externalId: string } + | { date: string; moodLoggedAt: Date } + > = entries.map((entry) => + entry.externalId + ? { source: entry.source, externalId: entry.externalId } + : { + date: moodDateKey(entry.moodLoggedAt, tz), + moodLoggedAt: entry.moodLoggedAt, + }, + ); + const existingRows = + orClauses.length > 0 + ? await prisma.moodEntry.findMany({ + where: { userId, OR: orClauses }, + select: { + id: true, + deletedAt: true, + source: true, + externalId: true, + date: true, + moodLoggedAt: true, + }, + }) + : []; + const byExternalKey = new Map(); + const byDateKey = new Map(); + for (const row of existingRows) { + if (row.externalId !== null) { + byExternalKey.set(`${row.source}::${row.externalId}`, row); + } + byDateKey.set(`${row.date}::${row.moodLoggedAt.toISOString()}`, row); + } + return { byExternalKey, byDateKey }; +} + +/** + * The upsert and the tag/factor links run in ONE transaction, matching the + * single-entry POST path. Split apart, an out-of-scale `ratedFactors` value + * (validated inside `createTagLinks`, i.e. AFTER the row committed) left a + * persisted mood entry with no links while the per-entry catch reported the + * entry as "skipped": the client believes nothing saved and a half-written + * row is on disk. + */ +async function writeEntry(args: { + userId: string; + tz: string; + entry: BulkMoodEntry; + date: string; + levelA: ReturnType; + context: z.infer | null | undefined; +}): Promise<{ id: string; droppedKeys: DroppedLinkKeys }> { + const { userId, tz, entry, date, levelA } = args; + const score = getScoreForMood(entry.mood); + // When the entry carries a source-stable `externalId`, dedup on the + // NULL-distinct `(userId, source, externalId)` key so a re-post with the + // same id (an iOS retry after a network hiccup, or a second adopt-on-pair + // backfill) updates the existing row in place instead of minting a + // duplicate when `moodLoggedAt` re-rounds / re-zones. Absent → the legacy + // `(userId, date, moodLoggedAt)` key. `source` carries a schema default of + // "MANUAL"; resolve it once so the probe, upsert key, and create write all + // agree on the value. + const resolvedSource = entry.source; + const probeWhere = entry.externalId + ? { + userId_source_externalId: { + userId, + source: resolvedSource, + externalId: entry.externalId, + }, + } + : { + userId_date_moodLoggedAt: { + userId, + date, + moodLoggedAt: entry.moodLoggedAt, + }, + }; + return prisma.$transaction(async (tx) => { + const upserted = await tx.moodEntry.upsert({ + where: probeWhere, + create: { + userId, + date, + tz, + mood: entry.mood, + score, + ...levelA, + tags: entry.tags ? JSON.stringify(entry.tags) : null, + note: null, + noteEncrypted: encryptNote(entry.note ?? null), + source: resolvedSource, + externalId: entry.externalId ?? null, + moodLoggedAt: entry.moodLoggedAt, + }, + update: { + // Last-writer-wins on the mood + tags + note triple. The iOS client + // only re-posts an existing entry when it has new data; the server + // trusts that decision. When the dedup key is `externalId`, also + // refresh `date` / `moodLoggedAt` so a re-zoned re-import lands the + // corrected wall-clock on the same row. + mood: entry.mood, + score, + ...levelA, + tags: entry.tags ? JSON.stringify(entry.tags) : null, + note: null, + noteEncrypted: encryptNote(entry.note ?? null), + ...(entry.externalId ? { date, moodLoggedAt: entry.moodLoggedAt } : {}), + }, + }); + + // Structured-tag links, mirroring the single-entry `createTagLinks` + // path. Additive + idempotent: `createTagLinks` resolves keys against the + // catalog (dropping unknown keys) and `skipDuplicates` on the join insert + // keeps a re-posted entry from minting duplicate links. Runs for both + // fresh and re-posted (upserted) rows so a backfill that adds tag keys on + // a second pass still lands them. Rated factors ride the same path; an + // out-of-scale rating throws `RatedFactorOutOfRangeError`, which inside + // the transaction rolls the mood row back with it, so a "skipped" result + // means nothing was written. The day context sits inside the same + // transaction as the row for the same reason. + await persistMoodContext(tx, upserted.id, userId, args.context); + + // The keys the catalog did not resolve are reported on this entry's + // result: "inserted" alone told a client its tags had landed. + let droppedKeys = NO_DROPPED_LINK_KEYS; + if ( + (entry.tagKeys && entry.tagKeys.length > 0) || + (entry.ratedFactors && entry.ratedFactors.length > 0) + ) { + droppedKeys = await createTagLinks( + upserted.id, + userId, + entry.tagKeys ?? [], + tx, + entry.ratedFactors ?? [], + ); + } + return { id: upserted.id, droppedKeys }; + }); +} + +/** + * P2002 = a concurrent batch raced the same NULL-distinct + * `(userId, source, externalId)` key and won. That is the plain duplicate the + * sequential probe would have classified; reporting it as "skipped" with a + * raw Prisma message told the client its row was dropped when it is in fact + * stored. Resolve the winning row's id so the caller's cursor advances, + * exactly like the intake bulk route. + */ +function isRacedDuplicate(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code: string }).code === "P2002" + ); +} + +async function ingestEntry(args: { + userId: string; + tz: string; + entry: BulkMoodEntry; + index: number; + existing: ExistingRows; + parseDimensions: ParseBulkDimensions; + tally: Tally; +}): Promise { + const { userId, tz, entry, index: i, tally } = args; + + // The `(userId, source, externalId)` dedup key only holds while the id is + // STABLE across client launches; a per-process value (an object + // description carrying a memory address) mints a fresh entry on every + // re-post. Per-entry refusal, never a whole-batch 422. + const unstable = + entry.externalId === undefined + ? null + : classifyExternalId(entry.externalId); + if (unstable) { + tally.unstableShapes.push(unstable); + tally.skipped.push({ index: i, reason: UNSTABLE_EXTERNAL_ID_REASON }); + tally.results.push({ + index: i, + status: "skipped", + reason: UNSTABLE_EXTERNAL_ID_REASON, + externalId: entry.externalId, + }); + return; + } + + // The context, parsed for THIS entry. A value outside the vocabulary skips + // this row and nothing else. + const parsedContext = + entry.context === undefined + ? { success: true as const, data: undefined } + : moodContextSchema.nullable().safeParse(entry.context); + if (!parsedContext.success) { + skip(tally, i, entry, "invalid_context"); + return; + } + + // The five dimensions, parsed for THIS entry. Out of range skips this row + // and nothing else. + const parsedDimensions = args.parseDimensions({ + a1: entry.a1, + a2: entry.a2, + a3: entry.a3, + a4: entry.a4, + a5: entry.a5, + }); + if (!parsedDimensions.success) { + skip(tally, i, entry, "invalid_dimensions"); + return; + } + + const date = moodDateKey(entry.moodLoggedAt, tz); + // The same resolution the single-entry path uses: pleasantness derived + // from the label unless the entry states its own, and the other four + // present only when this entry carried them. An omitted dimension is left + // alone rather than written as null, so a row that already carries + // hand-set values keeps them across a re-import. + const levelA = levelAForWrite(entry.mood, parsedDimensions.data); + + try { + // The response distinguishes "inserted" from "duplicate" via the batched + // pre-read (keyed exactly like the upsert key), so the per-entry probe + // here is a Map lookup, not a round trip. + const existing = + (entry.externalId + ? args.existing.byExternalKey.get( + `${entry.source}::${entry.externalId}`, + ) + : args.existing.byDateKey.get( + `${date}::${entry.moodLoggedAt.toISOString()}`, + )) ?? null; + + // Tombstone suppression: a soft-deleted match stays deleted. The + // user-facing DELETE route flips `deletedAt` so the `/api/sync/changes` + // feed surfaces the deletion to paired clients offline at delete time + // (see the MoodEntry schema note); an offline client's later re-post of + // the same entry is stale state, not a recreation, and must not resurrect + // the row. True no-op: no value churn, no `updatedAt` bump, no tag-link + // writes on the hidden row; report `duplicate` so the client checkpoints + // past it exactly like a live-row match. + if (existing?.deletedAt) { + tally.duplicates += 1; + tally.results.push({ + index: i, + status: "duplicate", + id: existing.id, + ...echoExternalId(entry), + }); + return; + } + + const written = await writeEntry({ + userId, + tz, + entry, + date, + levelA, + context: parsedContext.data, + }); + tally.droppedTagKeyCount += written.droppedKeys.droppedTagKeys.length; + tally.droppedFactorKeyCount += written.droppedKeys.droppedFactorKeys.length; + + if (existing) tally.duplicates += 1; + else tally.inserted += 1; + tally.results.push({ + index: i, + status: existing ? "duplicate" : "inserted", + id: written.id, + ...echoExternalId(entry), + ...droppedLinkKeysForWire(written.droppedKeys), + }); + } catch (err: unknown) { + if (isRacedDuplicate(err) && entry.externalId) { + const winner = await prisma.moodEntry.findFirst({ + where: { + userId, + // Same resolution as the upsert key: the schema defaults `source` + // to "MANUAL" at parse time. + source: entry.source, + externalId: entry.externalId, + }, + select: { id: true }, + }); + tally.duplicates += 1; + tally.results.push({ + index: i, + status: "duplicate", + ...(winner ? { id: winner.id } : {}), + externalId: entry.externalId, + }); + return; + } + const reason = + err instanceof Error ? err.message.slice(0, 120) : "upsert_failed"; + skip(tally, i, entry, reason); + } +} + +/** + * Refresh the rollup tier for every distinct day the batch touched. The bulk + * endpoint is an iOS one-shot backfill, so the batch can span many days; the + * set of `date` labels (the same per-row `moodDateKey` the insert used, so + * the rollup keys byte-identically to the stored `MoodEntry.date`) bounds the + * recompute count. Best-effort: a rollup failure must not surface as a 5xx. + */ +async function refreshTouchedRollups( + userId: string, + tz: string, + entries: BulkMoodEntry[], + results: EntryResult[], +): Promise { + const touchedLabels = new Set(); + for (let i = 0; i < entries.length; i++) { + const status = results[i]?.status; + if (status === "inserted" || status === "duplicate") { + touchedLabels.add(moodDateKey(entries[i].moodLoggedAt, tz)); + } + } + try { + await Promise.all( + Array.from(touchedLabels).map((label) => + recomputeMoodBucketsForEntry(userId, label), + ), + ); + } catch (rollupErr) { + annotate({ + meta: { + mood_rollup_write_failed: true, + mood_rollup_write_error: + rollupErr instanceof Error ? rollupErr.message : String(rollupErr), + }, + }); + } +} + +/** Write one parsed batch for `userId` and answer its per-entry envelope. */ +export async function ingestMoodBulk(input: { + userId: string; + tz: string; + entries: BulkMoodEntry[]; + ipAddress: string | null; + parseDimensions: ParseBulkDimensions; +}): Promise { + const { userId, tz, entries } = input; + const tally: Tally = { + results: [], + skipped: [], + inserted: 0, + duplicates: 0, + droppedTagKeyCount: 0, + droppedFactorKeyCount: 0, + unstableShapes: [], + }; + const existing = await readExistingRows(userId, tz, entries); + + for (let i = 0; i < entries.length; i++) { + await ingestEntry({ + userId, + tz, + entry: entries[i], + index: i, + existing, + parseDimensions: input.parseDimensions, + tally, + }); + } + + await auditLog("mood.bulk.ingest", { + userId, + ipAddress: input.ipAddress, + details: { + processed: entries.length, + inserted: tally.inserted, + duplicates: tally.duplicates, + skipped: tally.skipped.length, + }, + }); + + annotate({ + action: { name: "mood.bulk.ingest" }, + meta: { + processed: entries.length, + inserted: tally.inserted, + duplicates: tally.duplicates, + skipped: tally.skipped.length, + dropped_tag_keys: tally.droppedTagKeyCount, + dropped_factor_keys: tally.droppedFactorKeyCount, + ...(tally.unstableShapes.length > 0 + ? unstableExternalIdMeta("mood.bulk", tally.unstableShapes) + : {}), + }, + }); + + // Bust the per-user mood + achievements + analytics caches when at least + // one row landed so the next read picks up the ingested batch. Skipped / + // duplicate-only ingests are no-ops. + if (tally.inserted > 0) { + invalidateUserMood(userId); + } + + if (tally.inserted > 0 || tally.duplicates > 0) { + await refreshTouchedRollups(userId, tz, entries, tally.results); + } + + // 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 a mood drain advances it. + await markSyncCheckpoint(userId); + + return { + processed: entries.length, + inserted: tally.inserted, + duplicates: tally.duplicates, + skipped: tally.skipped, + entries: tally.results, + }; +}