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

refactor(import): move the JSON import writers out of their route

Browse Source
This commit is contained in:
Marc-André Bombeck committed 2026-10-23 14:51:51 +02:00
1 parent ed12625ab6
commit 11a7124d12
10 files changed
+560 -397
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/__tests__/baselines/source-function-lines.json
+2 -3
View File
@@ -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,
src/__tests__/baselines/version-stamp-comments.json
+2 -2
View File
@@ -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,
src/__tests__/external-id-stability-guard.test.ts
+5 -1
View File
@@ -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);
src/__tests__/measurement-write-plausibility-guard.test.ts
+4 -1
View File
@@ -73,7 +73,10 @@ const WRITERS: Record<string, Disposition> = {
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.
src/__tests__/measurement-write-tail-guard.test.ts
+1 -1
View File
@@ -53,7 +53,7 @@ const PROVIDER_TAIL = "runs afterProviderWrite";
*/
const WRITE_SITES: Record<string, string> = {
"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,
src/app/api/import/route.ts
+14 -389
View File
@@ -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<typeof moodEntrySchema>;
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<string, number>();
const reservedFallbackInstants = new Set<number>();
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",
src/lib/import/json/measurements.ts
+161
View File
@@ -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<typeof measurementTypeEnum>;
value: number;
unit: string;
measuredAt: Date;
glucoseContext?: z.infer<typeof glucoseContextEnum>;
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<ImportedMeasurements> {
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 };
}
src/lib/import/json/mood.ts
+253
View File
@@ -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<string, number>;
reserved: Set<number>;
}
async function writeMoodEntry(
userId: string,
e: JsonImportMoodEntry,
loggedAt: Date,
): Promise<void> {
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<boolean> {
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<void> {
// 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<void> {
const fallback: FallbackState = {
occurrences: new Map(),
reserved: new Set(),
};
for (const e of rows) {
await importMoodEntry(userId, e, fallback, tally);
if (tally.writeFailed) break;
}
}
src/lib/import/json/run.ts
+101
View File
@@ -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<Pick<JsonImportTally, "stats" | "writeFailed">> {
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 };
}
src/lib/import/json/tally.ts
+17
View File
@@ -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,
};
}