diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index a853ae209..3fcd96101 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", - "count": 379, - "total": 133508, + "count": 378, + "total": 133144, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -23,7 +23,6 @@ "app/api/environment/route.ts :: apiHandler(arg)": 201, "app/api/export/health-record/route.ts :: apiHandler(arg)": 241, "app/api/import/csv/route.ts :: apiHandler(arg)": 407, - "app/api/import/route.ts :: apiHandler(arg)": 364, "app/api/ingest/medication/route.ts :: apiHandler(arg)": 267, "app/api/insights/cards/route.ts :: apiHandler(arg)": 211, "app/api/insights/comprehensive/route.ts :: buildComprehensiveResponse": 505, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index b60f0262b..19ea6ce10 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": 2914, - "total": 9848, + "total": 9844, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -284,7 +284,7 @@ "app/api/illness/episodes/route.ts": 1, "app/api/import/__tests__/route.test.ts": 3, "app/api/import/csv/__tests__/route.test.ts": 2, - "app/api/import/route.ts": 8, + "app/api/import/route.ts": 4, "app/api/ingest/medication/__tests__/route.test.ts": 1, "app/api/ingest/medication/route.ts": 7, "app/api/insights/biomarker-assessment/route.ts": 1, diff --git a/src/__tests__/external-id-stability-guard.test.ts b/src/__tests__/external-id-stability-guard.test.ts index 8b3362015..4440d3783 100644 --- a/src/__tests__/external-id-stability-guard.test.ts +++ b/src/__tests__/external-id-stability-guard.test.ts @@ -409,7 +409,11 @@ describe("the batch surfaces refuse per entry, never whole-batch", () => { "lib/workouts/batch-ingest/ingest.ts", ], "app/api/cycle/day-logs/bulk/route.ts": [], - "app/api/import/route.ts": [], + "app/api/import/route.ts": [ + "lib/import/json/measurements.ts", + "lib/import/json/mood.ts", + "lib/import/json/run.ts", + ], }; const BATCH_ROUTES = Object.keys(BATCH_SURFACES); diff --git a/src/__tests__/measurement-write-plausibility-guard.test.ts b/src/__tests__/measurement-write-plausibility-guard.test.ts index 12c9de285..63c16b88a 100644 --- a/src/__tests__/measurement-write-plausibility-guard.test.ts +++ b/src/__tests__/measurement-write-plausibility-guard.test.ts @@ -73,7 +73,10 @@ const WRITERS: Record = { kind: "upstream", validatedIn: "lib/validations/measurement.ts", }, - "app/api/import/route.ts": { kind: "validator" }, + "lib/import/json/measurements.ts": { + kind: "upstream", + validatedIn: "app/api/import/route.ts", + }, // The batch ingest's set-based insert writes rows that already passed // `validateMeasurementRange` per entry in its first pass; everything else // goes through the reconciler, which gates itself. diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 9a7f671a4..eaa3757a6 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -53,7 +53,7 @@ const PROVIDER_TAIL = "runs afterProviderWrite"; */ const WRITE_SITES: Record = { "app/api/import/csv/route.ts": TAIL, - "app/api/import/route.ts": TAIL, + "lib/import/json/measurements.ts": TAIL_VIA("lib/import/json/run.ts"), "app/api/measurements/[id]/route.ts": TAIL, "app/api/measurements/bulk-delete/route.ts": TAIL, "app/api/measurements/by-external-ids/route.ts": TAIL, diff --git a/src/app/api/import/route.ts b/src/app/api/import/route.ts index a1b46f9d1..8931865f0 100644 --- a/src/app/api/import/route.ts +++ b/src/app/api/import/route.ts @@ -1,7 +1,10 @@ -import { prisma } from "@/lib/db"; +/** + * `POST /api/import` — import user data (JSON format matching the export + * structure). The route authenticates, meters and parses the file; the + * writers live in `src/lib/import/json/`. + */ import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; -import { auditLog } from "@/lib/auth/audit"; import { apiError, apiSuccess, @@ -13,7 +16,6 @@ import { import { checkRateLimit } from "@/lib/rate-limit"; import { NextRequest } from "next/server"; import { z } from "zod/v4"; -import { createHash } from "node:crypto"; import { validateMeasurementRange, @@ -21,27 +23,12 @@ import { glucoseContextEnum, } from "@/lib/validations/measurement"; import { - ENTRY_INSTANT_CLOCK_SKEW_MS, isPlausibleEntryInstant, validateEntryInstant, } from "@/lib/validations/entry-instant"; -import { - classifyExternalId, - unstableExternalIdMeta, - type UnstableExternalIdShape, -} from "@/lib/validations/external-id"; -import { encryptNote } from "@/lib/crypto/note-cipher"; -import { recomputeUserMoodRollups } from "@/lib/rollups/mood-rollups"; -import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; -import type { MeasurementType } from "@/generated/prisma/client"; -import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; -import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; -import { isP2002, isP2025 } from "@/lib/prisma-errors"; -import { moodDateKey } from "@/lib/mood/date-key"; -import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; -import { deriveA1 } from "@/lib/mood/level-a"; -import { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; import { isoDateTime } from "@/lib/validations/datetime"; +import { fallbackMoodDateStart } from "@/lib/import/json/mood"; +import { runJsonImport } from "@/lib/import/json/run"; // Derived from canonical enum so round-trip export → import covers every // type. Previous hardcoded subset silently dropped 4 of 11 types @@ -70,7 +57,7 @@ const measurementSchema = z // Stability is enforced per entry in the import loop, not at the // field: a schema-level refusal would fail the whole file on one bad // row. - // @external-id-checked-per-entry: src/app/api/import/route.ts + // @external-id-checked-per-entry: src/lib/import/json/measurements.ts externalId: z.string().min(1).max(120).optional(), }) .refine( @@ -78,16 +65,6 @@ const measurementSchema = z "Wert ausserhalb des plausiblen Bereichs", ); -function fallbackMoodDateStart(date: string): Date | null { - const [year, month, day] = date.split("-").map(Number); - if (!year || !month || !day) return null; - const instant = zonedWallClockToUtc( - { year, month, day, hour: 0, minute: 0 }, - DEFAULT_TIMEZONE, - ); - return moodDateKey(instant, DEFAULT_TIMEZONE) === date ? instant : null; -} - const moodEntrySchema = z .object({ date: z.string().regex(/^\d{4}-\d{2}-\d{2}$/), @@ -106,7 +83,7 @@ const moodEntrySchema = z // Stability is enforced per entry in the import loop, not at the // field: a schema-level refusal would fail the whole file on one bad // row. - // @external-id-checked-per-entry: src/app/api/import/route.ts + // @external-id-checked-per-entry: src/lib/import/json/mood.ts externalId: z.string().min(1).max(120).optional(), }) .superRefine((entry, ctx) => { @@ -121,42 +98,11 @@ const moodEntrySchema = z } }); -type MoodEntryInput = z.infer; - -function fallbackMoodIdentity(entry: MoodEntryInput): string { - return JSON.stringify([ - entry.date, - entry.mood, - entry.score, - entry.tags ?? null, - entry.externalId ?? null, - ]); -} - -function stableFallbackMoodInstant( - entry: MoodEntryInput, - occurrence: number, - probe: number, -): Date { - const digest = createHash("sha256") - .update(`${fallbackMoodIdentity(entry)}:${occurrence}`) - .digest(); - const initialOffset = digest.readUInt32BE(0) % ENTRY_INSTANT_CLOCK_SKEW_MS; - const offset = (initialOffset + probe) % ENTRY_INSTANT_CLOCK_SKEW_MS; - const dateStart = fallbackMoodDateStart(entry.date); - if (!dateStart) throw new Error("Invalid mood date"); - return new Date(dateStart.getTime() + offset); -} - const importSchema = z.object({ measurements: z.array(measurementSchema).max(10000).optional(), moodEntries: z.array(moodEntrySchema).max(10000).optional(), }); -/** - * Import user data (JSON format matching export structure). - * Deduplicates via unique constraints — existing records are skipped. - */ export const POST = apiHandler(async (request: NextRequest) => { const { user } = await requireAuth(); annotate({ action: { name: "import.upload" } }); @@ -183,334 +129,13 @@ export const POST = apiHandler(async (request: NextRequest) => { ); } - const userId = user.id; - const data = parsed.data; - const stats = { measurements: 0, moodEntries: 0, skipped: 0 }; - const unstableShapes: UnstableExternalIdShape[] = []; - let writeFailed = false; - - // Import measurements - // v1.4.39.1 — track each (type, measuredAt) we wrote so we can re- - // fold the persistent rollup tier at the end. Pre-fix, the import - // path bypassed the rollup write hook entirely, so a CSV / JSON - // restore left the user's `source=rollup` chart paths under-counting - // the imported days until the next worker boot re-ran the backfill. - const touchedMeasurements: Array<{ - type: MeasurementType; - measuredAt: Date; - }> = []; - const insertedMeasurements: Array<{ - id: string; - type: MeasurementType; - measuredAt: Date; - }> = []; - if (data.measurements?.length) { - for (const m of data.measurements) { - // The `(userId, type, source, externalId)` upsert key is only - // idempotent while the id is STABLE; an id that rotates between - // exports re-imports as a fresh row every time. Skipped per entry, - // like every other rejected row in this importer. - const unstable = - m.externalId === undefined ? null : classifyExternalId(m.externalId); - if (unstable) { - unstableShapes.push(unstable); - stats.skipped++; - continue; - } - try { - // `measuredAt` is already a `Date` (validateEntryInstant transform). - const measuredAt = m.measuredAt; - if (m.externalId) { - const notesEncrypted = encryptNote(m.notes || null); - // INSERT ... RETURNING is the only insertion proof. A concurrent - // importer can win this key after any existence probe, so attempt - // the insert first and update only when the statement returns no row. - const created = await prisma.measurement.createManyAndReturn({ - data: [ - { - userId, - type: m.type, - value: m.value, - unit: m.unit, - source: "IMPORT", - externalId: m.externalId, - measuredAt, - notes: null, - notesEncrypted, - glucoseContext: m.glucoseContext ?? null, - }, - ], - skipDuplicates: true, - select: { id: true, type: true, measuredAt: true }, - }); - const inserted = created[0]; - if (inserted) { - insertedMeasurements.push(inserted); - } else { - // A pre-existing row or a raced insert owns the key. Preserve - // idempotent re-import semantics without classifying this update - // as an arrival. - try { - await prisma.measurement.update({ - where: { - userId_type_source_externalId: { - userId, - type: m.type, - source: "IMPORT", - externalId: m.externalId, - }, - }, - data: { - value: m.value, - unit: m.unit, - measuredAt, - notes: null, - notesEncrypted, - glucoseContext: m.glucoseContext ?? null, - }, - }); - } catch (err) { - if (isP2025(err)) { - stats.skipped++; - continue; - } - throw err; - } - } - } else { - const inserted = await prisma.measurement.create({ - data: { - userId, - type: m.type, - value: m.value, - unit: m.unit, - source: "IMPORT", - measuredAt, - notes: null, - notesEncrypted: encryptNote(m.notes || null), - glucoseContext: m.glucoseContext ?? null, - }, - select: { id: true, type: true, measuredAt: true }, - }); - insertedMeasurements.push(inserted); - } - touchedMeasurements.push({ - type: m.type as MeasurementType, - measuredAt, - }); - stats.measurements++; - } catch (err) { - if (isP2002(err)) { - stats.skipped++; - continue; - } - writeFailed = true; - break; - } - } - } - - void emitInsertedMeasurementArrivals( - userId, - insertedMeasurements, - "json_import", - ).catch(() => {}); - void maybeEnqueueMorningRefresh( - userId, - insertedMeasurements - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt), - ).catch(() => {}); - - // Import mood entries - if (!writeFailed && data.moodEntries?.length) { - const fallbackOccurrences = new Map(); - const reservedFallbackInstants = new Set(); - for (const e of data.moodEntries) { - // Same stability floor as the measurement loop above — the mood - // upsert keys on `(userId, source, externalId)`. - const unstableMood = - e.externalId === undefined ? null : classifyExternalId(e.externalId); - if (unstableMood) { - unstableShapes.push(unstableMood); - stats.skipped++; - continue; - } - const identity = e.loggedAt ? null : fallbackMoodIdentity(e); - const occurrence = - identity === null ? 0 : (fallbackOccurrences.get(identity) ?? 0); - if (identity !== null) { - fallbackOccurrences.set(identity, occurrence + 1); - } - - for (let probe = 0; probe < ENTRY_INSTANT_CLOCK_SKEW_MS; probe++) { - const loggedAt = - e.loggedAt ?? stableFallbackMoodInstant(e, occurrence, probe); - if (!e.loggedAt && reservedFallbackInstants.has(loggedAt.getTime())) { - continue; - } - try { - if (e.externalId) { - // v1.12.1 — idempotent re-import keyed on the source-stable id. - // A second import of the same export updates the row in place - // rather than skipping it as a duplicate, so an upstream edit - // (re-scored mood, added tag) is reflected. - await prisma.moodEntry.upsert({ - where: { - userId_source_externalId: { - userId, - source: "IMPORT", - externalId: e.externalId, - }, - }, - update: { - date: e.date, - mood: e.mood, - score: e.score, - // v1.37 — pleasantness derived from the row's own label, not - // from the file's `score`. This is the one writer whose score - // comes out of the file rather than out of the label, so the - // two can disagree; the column that records a self-assessment - // has to agree with the label the user actually picked. - moodA1: deriveA1(e.mood), - tags: e.tags || null, - moodLoggedAt: loggedAt, - }, - create: { - userId, - date: e.date, - mood: e.mood, - score: e.score, - moodA1: deriveA1(e.mood), - tags: e.tags || null, - source: "IMPORT", - externalId: e.externalId, - moodLoggedAt: loggedAt, - }, - }); - } else { - await prisma.moodEntry.create({ - data: { - userId, - date: e.date, - mood: e.mood, - score: e.score, - // See the upsert arm above: derived from the label, because - // an imported file's `score` is its own field. - moodA1: deriveA1(e.mood), - tags: e.tags || null, - source: "IMPORT", - moodLoggedAt: loggedAt, - }, - }); - } - if (!e.loggedAt) { - reservedFallbackInstants.add(loggedAt.getTime()); - } - stats.moodEntries++; - break; - } catch (err) { - if (!isP2002(err)) { - writeFailed = true; - break; - } - if (e.loggedAt) { - stats.skipped++; - break; - } - if (!e.externalId) { - try { - const existing = await prisma.moodEntry.findUnique({ - where: { - userId_date_moodLoggedAt: { - userId, - date: e.date, - moodLoggedAt: loggedAt, - }, - }, - select: { - source: true, - mood: true, - score: true, - tags: true, - externalId: true, - }, - }); - if ( - existing?.source === "IMPORT" && - existing.mood === e.mood && - existing.score === e.score && - existing.tags === (e.tags || null) && - (existing.externalId ?? null) === null - ) { - reservedFallbackInstants.add(loggedAt.getTime()); - stats.skipped++; - break; - } - } catch { - writeFailed = true; - break; - } - } - if (probe === ENTRY_INSTANT_CLOCK_SKEW_MS - 1) { - writeFailed = true; - } - } - } - if (writeFailed) break; - } - } - - // v1.4.39 — one bounded re-fold per import is cheaper than - // firing N per-row hooks. The mood rollup tier is keyed on the - // user + day; the fold materialises every touched day from the - // entries we just wrote. Best-effort: an importer hiccup must not - // surface as a 5xx, the rollup is a cache tier. - if (stats.moodEntries > 0) { - try { - await recomputeUserMoodRollups(userId, { granularities: ["DAY"] }); - } catch (err) { - annotate({ - meta: { - mood_rollup_import_failed: true, - mood_rollup_import_error: - err instanceof Error ? err.message : String(err), - }, - }); - } - } - - // Refresh the persistent measurement rollup table for every distinct - // (type, day) the import touched (collapsed, so a 10 000-row restore pays - // at most ~N recomputes) and re-warm the touched types' status - // assessments. Best-effort: a populator hiccup never fails the importer. - await afterMeasurementMutation(userId, touchedMeasurements, { - // A file the person chose to import: a deliberate act, narrated like a - // hand entry rather than held to the sync gap. - source: "manual", - label: "import", + const { stats, writeFailed } = await runJsonImport({ + userId: user.id, + measurements: parsed.data.measurements, + moodEntries: parsed.data.moodEntries, + ipAddress: getClientIp(request), }); - annotate({ - meta: { - import_stats: stats, - ...(unstableShapes.length > 0 - ? unstableExternalIdMeta("import.json", unstableShapes) - : {}), - }, - }); - - try { - await auditLog("import.upload", { - userId, - ipAddress: getClientIp(request), - details: stats, - }); - } catch (err) { - if (!writeFailed) throw err; - annotate({ meta: { import_audit_failed: true } }); - } - if (writeFailed) { return apiError("Import write failed", 503, { errorCode: "import.write_failed", diff --git a/src/lib/import/json/measurements.ts b/src/lib/import/json/measurements.ts new file mode 100644 index 000000000..1e57608ea --- /dev/null +++ b/src/lib/import/json/measurements.ts @@ -0,0 +1,161 @@ +/** + * The measurement rows of a JSON import (`POST /api/import`). + * + * The route's schema has already bounded each row (plausible range, entry + * instant). With an external id a row is inserted or, when the key is taken, + * updated in place, so a re-import of the same export is idempotent; without + * one it is a first-write-wins create with a NULL external id, which does + * duplicate on re-import. + */ +import type { z } from "zod/v4"; + +import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; +import type { + glucoseContextEnum, + measurementTypeEnum, +} from "@/lib/validations/measurement"; +import { classifyExternalId } from "@/lib/validations/external-id"; +import { encryptNote } from "@/lib/crypto/note-cipher"; +import { isP2002, isP2025 } from "@/lib/prisma-errors"; + +import type { JsonImportTally } from "./tally"; + +export interface JsonImportMeasurement { + type: z.infer; + value: number; + unit: string; + measuredAt: Date; + glucoseContext?: z.infer; + source?: string; + notes?: string; + externalId?: string; +} + +export interface ImportedMeasurements { + /** + * Each `(type, measuredAt)` written, so the persistent rollup tier is + * re-folded at the end. The import path once bypassed the rollup write + * hook entirely, so a restore left the `source=rollup` chart paths + * under-counting the imported days until the next worker boot re-ran the + * backfill. + */ + touched: Array<{ type: MeasurementType; measuredAt: Date }>; + /** The rows this import inserted (not the ones it updated in place). */ + inserted: Array<{ id: string; type: MeasurementType; measuredAt: Date }>; +} + +/** + * Write `rows` for `userId`, counting into `tally`. Stops at the first + * write that fails for any reason other than a duplicate, and marks the + * tally failed. + */ +export async function importJsonMeasurements( + userId: string, + rows: JsonImportMeasurement[], + tally: JsonImportTally, +): Promise { + const touched: ImportedMeasurements["touched"] = []; + const inserted: ImportedMeasurements["inserted"] = []; + for (const m of rows) { + // The `(userId, type, source, externalId)` upsert key is only + // idempotent while the id is STABLE; an id that rotates between + // exports re-imports as a fresh row every time. Skipped per entry, + // like every other rejected row in this importer. + const unstable = + m.externalId === undefined ? null : classifyExternalId(m.externalId); + if (unstable) { + tally.unstableShapes.push(unstable); + tally.stats.skipped++; + continue; + } + try { + // `measuredAt` is already a `Date` (validateEntryInstant transform). + const measuredAt = m.measuredAt; + if (m.externalId) { + const notesEncrypted = encryptNote(m.notes || null); + // INSERT ... RETURNING is the only insertion proof. A concurrent + // importer can win this key after any existence probe, so attempt + // the insert first and update only when the statement returns no row. + const created = await prisma.measurement.createManyAndReturn({ + data: [ + { + userId, + type: m.type, + value: m.value, + unit: m.unit, + source: "IMPORT", + externalId: m.externalId, + measuredAt, + notes: null, + notesEncrypted, + glucoseContext: m.glucoseContext ?? null, + }, + ], + skipDuplicates: true, + select: { id: true, type: true, measuredAt: true }, + }); + const row = created[0]; + if (row) { + inserted.push(row); + } else { + // A pre-existing row or a raced insert owns the key. Preserve + // idempotent re-import semantics without classifying this update + // as an arrival. + try { + await prisma.measurement.update({ + where: { + userId_type_source_externalId: { + userId, + type: m.type, + source: "IMPORT", + externalId: m.externalId, + }, + }, + data: { + value: m.value, + unit: m.unit, + measuredAt, + notes: null, + notesEncrypted, + glucoseContext: m.glucoseContext ?? null, + }, + }); + } catch (err) { + if (isP2025(err)) { + tally.stats.skipped++; + continue; + } + throw err; + } + } + } else { + const row = await prisma.measurement.create({ + data: { + userId, + type: m.type, + value: m.value, + unit: m.unit, + source: "IMPORT", + measuredAt, + notes: null, + notesEncrypted: encryptNote(m.notes || null), + glucoseContext: m.glucoseContext ?? null, + }, + select: { id: true, type: true, measuredAt: true }, + }); + inserted.push(row); + } + touched.push({ type: m.type as MeasurementType, measuredAt }); + tally.stats.measurements++; + } catch (err) { + if (isP2002(err)) { + tally.stats.skipped++; + continue; + } + tally.writeFailed = true; + break; + } + } + return { touched, inserted }; +} diff --git a/src/lib/import/json/mood.ts b/src/lib/import/json/mood.ts new file mode 100644 index 000000000..675292a43 --- /dev/null +++ b/src/lib/import/json/mood.ts @@ -0,0 +1,253 @@ +/** + * The mood rows of a JSON import (`POST /api/import`). + * + * A row with an external id upserts on `(userId, source, externalId)`. A row + * without a logged instant gets a stable fallback instant inside its day, + * derived from the row's content and its occurrence in the file, so the same + * file lands on the same instants every time and twin rows land on distinct + * ones. + */ +import { createHash } from "node:crypto"; + +import { prisma } from "@/lib/db"; +import { ENTRY_INSTANT_CLOCK_SKEW_MS } from "@/lib/validations/entry-instant"; +import { classifyExternalId } from "@/lib/validations/external-id"; +import { isP2002 } from "@/lib/prisma-errors"; +import { moodDateKey } from "@/lib/mood/date-key"; +import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; +import { deriveA1 } from "@/lib/mood/level-a"; +import { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; + +import type { JsonImportTally } from "./tally"; + +export interface JsonImportMoodEntry { + date: string; + mood: "SUPER_GUT" | "GUT" | "OKAY" | "SCHLECHT" | "LAUSIG"; + score: number; + tags?: string; + loggedAt?: Date; + externalId?: string; +} + +/** + * The start of a `YYYY-MM-DD` day in the default zone, or null when the + * string does not name a real day. The route's schema uses it to bound a row + * without a logged instant. + */ +export function fallbackMoodDateStart(date: string): Date | null { + const [year, month, day] = date.split("-").map(Number); + if (!year || !month || !day) return null; + const instant = zonedWallClockToUtc( + { year, month, day, hour: 0, minute: 0 }, + DEFAULT_TIMEZONE, + ); + return moodDateKey(instant, DEFAULT_TIMEZONE) === date ? instant : null; +} + +function fallbackMoodIdentity(entry: JsonImportMoodEntry): string { + return JSON.stringify([ + entry.date, + entry.mood, + entry.score, + entry.tags ?? null, + entry.externalId ?? null, + ]); +} + +function stableFallbackMoodInstant( + entry: JsonImportMoodEntry, + occurrence: number, + probe: number, +): Date { + const digest = createHash("sha256") + .update(`${fallbackMoodIdentity(entry)}:${occurrence}`) + .digest(); + const initialOffset = digest.readUInt32BE(0) % ENTRY_INSTANT_CLOCK_SKEW_MS; + const offset = (initialOffset + probe) % ENTRY_INSTANT_CLOCK_SKEW_MS; + const dateStart = fallbackMoodDateStart(entry.date); + if (!dateStart) throw new Error("Invalid mood date"); + return new Date(dateStart.getTime() + offset); +} + +/** The fallback instants this import has handed out, and how often each row content occurred. */ +interface FallbackState { + occurrences: Map; + reserved: Set; +} + +async function writeMoodEntry( + userId: string, + e: JsonImportMoodEntry, + loggedAt: Date, +): Promise { + if (e.externalId) { + // Idempotent re-import keyed on the source-stable id. A second import + // of the same export updates the row in place rather than skipping it as + // a duplicate, so an upstream edit (re-scored mood, added tag) is + // reflected. + await prisma.moodEntry.upsert({ + where: { + userId_source_externalId: { + userId, + source: "IMPORT", + externalId: e.externalId, + }, + }, + update: { + date: e.date, + mood: e.mood, + score: e.score, + // Pleasantness derived from the row's own label, not from the file's + // `score`. This is the one writer whose score comes out of the file + // rather than out of the label, so the two can disagree; the column + // that records a self-assessment has to agree with the label the + // user actually picked. + moodA1: deriveA1(e.mood), + tags: e.tags || null, + moodLoggedAt: loggedAt, + }, + create: { + userId, + date: e.date, + mood: e.mood, + score: e.score, + moodA1: deriveA1(e.mood), + tags: e.tags || null, + source: "IMPORT", + externalId: e.externalId, + moodLoggedAt: loggedAt, + }, + }); + } else { + await prisma.moodEntry.create({ + data: { + userId, + date: e.date, + mood: e.mood, + score: e.score, + // See the upsert arm above: derived from the label, because an + // imported file's `score` is its own field. + moodA1: deriveA1(e.mood), + tags: e.tags || null, + source: "IMPORT", + moodLoggedAt: loggedAt, + }, + }); + } +} + +/** + * Whether the row already stored at `loggedAt` is this very fallback row + * from an earlier import of the same file. Throws when the probe fails. + */ +async function sameImportedRow( + userId: string, + e: JsonImportMoodEntry, + loggedAt: Date, +): Promise { + const existing = await prisma.moodEntry.findUnique({ + where: { + userId_date_moodLoggedAt: { + userId, + date: e.date, + moodLoggedAt: loggedAt, + }, + }, + select: { + source: true, + mood: true, + score: true, + tags: true, + externalId: true, + }, + }); + return ( + existing?.source === "IMPORT" && + existing.mood === e.mood && + existing.score === e.score && + existing.tags === (e.tags || null) && + (existing.externalId ?? null) === null + ); +} + +async function importMoodEntry( + userId: string, + e: JsonImportMoodEntry, + fallback: FallbackState, + tally: JsonImportTally, +): Promise { + // Same stability floor as the measurement rows: the mood upsert keys on + // `(userId, source, externalId)`. + const unstableMood = + e.externalId === undefined ? null : classifyExternalId(e.externalId); + if (unstableMood) { + tally.unstableShapes.push(unstableMood); + tally.stats.skipped++; + return; + } + const identity = e.loggedAt ? null : fallbackMoodIdentity(e); + const occurrence = + identity === null ? 0 : (fallback.occurrences.get(identity) ?? 0); + if (identity !== null) { + fallback.occurrences.set(identity, occurrence + 1); + } + + for (let probe = 0; probe < ENTRY_INSTANT_CLOCK_SKEW_MS; probe++) { + const loggedAt = + e.loggedAt ?? stableFallbackMoodInstant(e, occurrence, probe); + if (!e.loggedAt && fallback.reserved.has(loggedAt.getTime())) { + continue; + } + try { + await writeMoodEntry(userId, e, loggedAt); + if (!e.loggedAt) { + fallback.reserved.add(loggedAt.getTime()); + } + tally.stats.moodEntries++; + return; + } catch (err) { + if (!isP2002(err)) { + tally.writeFailed = true; + return; + } + if (e.loggedAt) { + tally.stats.skipped++; + return; + } + if (!e.externalId) { + try { + if (await sameImportedRow(userId, e, loggedAt)) { + fallback.reserved.add(loggedAt.getTime()); + tally.stats.skipped++; + return; + } + } catch { + tally.writeFailed = true; + return; + } + } + if (probe === ENTRY_INSTANT_CLOCK_SKEW_MS - 1) { + tally.writeFailed = true; + } + } + } +} + +/** + * Write `rows` for `userId`, counting into `tally`. Stops at the first row + * whose write failed for a reason other than a duplicate. + */ +export async function importJsonMoodEntries( + userId: string, + rows: JsonImportMoodEntry[], + tally: JsonImportTally, +): Promise { + const fallback: FallbackState = { + occurrences: new Map(), + reserved: new Set(), + }; + for (const e of rows) { + await importMoodEntry(userId, e, fallback, tally); + if (tally.writeFailed) break; + } +} diff --git a/src/lib/import/json/run.ts b/src/lib/import/json/run.ts new file mode 100644 index 000000000..1f965c4a8 --- /dev/null +++ b/src/lib/import/json/run.ts @@ -0,0 +1,101 @@ +/** + * One JSON import (`POST /api/import`): the measurement rows, then the mood + * rows, then the rollup and status tails, the wide event and the audit row. + * Deduplicates via unique constraints: existing records are skipped or, with + * an external id, updated in place. + */ +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import { unstableExternalIdMeta } from "@/lib/validations/external-id"; +import { recomputeUserMoodRollups } from "@/lib/rollups/mood-rollups"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; +import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; +import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; + +import { + importJsonMeasurements, + type JsonImportMeasurement, +} from "./measurements"; +import { importJsonMoodEntries, type JsonImportMoodEntry } from "./mood"; +import { newJsonImportTally, type JsonImportTally } from "./tally"; + +export async function runJsonImport(input: { + userId: string; + measurements: JsonImportMeasurement[] | undefined; + moodEntries: JsonImportMoodEntry[] | undefined; + ipAddress: string | null; +}): Promise> { + const { userId } = input; + const tally = newJsonImportTally(); + + const written = input.measurements?.length + ? await importJsonMeasurements(userId, input.measurements, tally) + : { touched: [], inserted: [] }; + + void emitInsertedMeasurementArrivals( + userId, + written.inserted, + "json_import", + ).catch(() => {}); + void maybeEnqueueMorningRefresh( + userId, + written.inserted + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt), + ).catch(() => {}); + + if (!tally.writeFailed && input.moodEntries?.length) { + await importJsonMoodEntries(userId, input.moodEntries, tally); + } + + // One bounded re-fold per import is cheaper than firing N per-row hooks. + // The mood rollup tier is keyed on the user + day; the fold materialises + // every touched day from the entries just written. Best-effort: an + // importer hiccup must not surface as a 5xx, the rollup is a cache tier. + if (tally.stats.moodEntries > 0) { + try { + await recomputeUserMoodRollups(userId, { granularities: ["DAY"] }); + } catch (err) { + annotate({ + meta: { + mood_rollup_import_failed: true, + mood_rollup_import_error: + err instanceof Error ? err.message : String(err), + }, + }); + } + } + + // Refresh the persistent measurement rollup table for every distinct + // (type, day) the import touched (collapsed, so a 10 000-row restore pays + // at most ~N recomputes) and re-warm the touched types' status + // assessments. Best-effort: a populator hiccup never fails the importer. + await afterMeasurementMutation(userId, written.touched, { + // A file the person chose to import: a deliberate act, narrated like a + // hand entry rather than held to the sync gap. + source: "manual", + label: "import", + }); + + annotate({ + meta: { + import_stats: tally.stats, + ...(tally.unstableShapes.length > 0 + ? unstableExternalIdMeta("import.json", tally.unstableShapes) + : {}), + }, + }); + + try { + await auditLog("import.upload", { + userId, + ipAddress: input.ipAddress, + details: tally.stats, + }); + } catch (err) { + if (!tally.writeFailed) throw err; + annotate({ meta: { import_audit_failed: true } }); + } + + return { stats: tally.stats, writeFailed: tally.writeFailed }; +} diff --git a/src/lib/import/json/tally.ts b/src/lib/import/json/tally.ts new file mode 100644 index 000000000..243150f06 --- /dev/null +++ b/src/lib/import/json/tally.ts @@ -0,0 +1,17 @@ +import type { UnstableExternalIdShape } from "@/lib/validations/external-id"; + +/** The running counts of one JSON import, shared by its row writers. */ +export interface JsonImportTally { + stats: { measurements: number; moodEntries: number; skipped: number }; + unstableShapes: UnstableExternalIdShape[]; + /** A write failed for a reason other than a duplicate; the import stops. */ + writeFailed: boolean; +} + +export function newJsonImportTally(): JsonImportTally { + return { + stats: { measurements: 0, moodEntries: 0, skipped: 0 }, + unstableShapes: [], + writeFailed: false, + }; +}