/** * Insert many measurements a hundred rows to a statement, as column arrays. * * Why not `createMany`. Prisma compiles and caches a query plan per query * shape, and a `createMany` of N rows is its own shape: measured, one * 1 000-row `createMany` of measurements leaves about 145 MB of retained * heap behind, a 2 000-row one about 290 MB, and the smaller last batch of a * restore adds its own plan on top. In a 1 GB container that alone took a * restore of 1.25 million measurements past the heap limit (#1031). One * `INSERT … SELECT FROM unnest(…)` with an array per column is one small * statement shape whatever the number of rows, and about four times faster. * * The Apple Health export import writes its spot rows through the same * statement (`insertNewMeasurementRows`), for the same reason. There the cost * was per call rather than per shape: every `createManyAndReturn` of a * 650-row flush left about 5 MB behind until roughly two dozen calls had * accumulated, so a 300 000-record import settled at about 53 MB of retained * heap and a run of back-to-back flushes reached well over 100 MB. Through * this statement the same import holds a few megabytes. * * Why no more than {@link MEASUREMENT_INSERT_ROWS_PER_STATEMENT} rows per * statement, whatever the caller hands over. Prisma's debug logger keeps the * arguments of its last hundred calls in memory whether or not logging is * switched on, and both the client and the Postgres adapter log every * statement together with its parameters. So the parameters of about the last * hundred statements stay reachable after they ran, as a fresh copy of every * value. At 1 000 rows a statement that held about 60 MB for the whole of a * restore's measurement write (#1031, measured on 1.8 million Apple Health * readings), next to an app whose heap limit in the default 1 GB container is * 524 MB. At 100 rows it is about 7 MB. The extra statements cost time: on a * fast host the same write took 94 s instead of 76 s, a price paid once per * restore, where the memory was held for all of it. * * `COLUMNS` maps every field of `MeasurementInsertRow` to its column and * type. It is typed as a complete record over the row's keys, so a field * added to the row without a column here does not compile, and the restore * builds its rows as `MeasurementInsertRow`, so a field it starts writing has * to be added to the row first. */ import { randomInt } from "node:crypto"; import { hostname } from "node:os"; import type { MeasurementType, Prisma, PrismaClient, } from "@/generated/prisma/client"; /** One measurement as the restore writes it. */ export interface MeasurementInsertRow { id: string; userId: string; type: string; value: number; valueMin: number | null; valueMax: number | null; unit: string; source: string; measuredAt: Date; notes: string | null; notesEncrypted: Uint8Array | null; externalId: string | null; externalSourceVersion: string | null; aggregationProvenance: string | null; glucoseContext: string | null; sleepStage: string | null; rhythmClassification: string | null; deviceType: string | null; /** The token an EXTERNAL row came from (#1272); null when absent. */ ingestTokenId?: string | null; syncVersion: number; deletedAt: Date | null; createdAt?: Date; updatedAt?: Date; } type Kind = "text" | "float8" | "int4" | "timestamp" | "bytea" | `enum:${string}`; /** Column and element type per field. Complete by construction. */ const COLUMNS: { [K in keyof MeasurementInsertRow]-?: [string, Kind] } = { id: ["id", "text"], userId: ["user_id", "text"], type: ["type", "enum:measurement_type"], value: ["value", "float8"], valueMin: ["value_min", "float8"], valueMax: ["value_max", "float8"], unit: ["unit", "text"], source: ["source", "enum:measurement_source"], measuredAt: ["measured_at", "timestamp"], notes: ["notes", "text"], notesEncrypted: ["notes_encrypted", "bytea"], externalId: ["external_id", "text"], externalSourceVersion: ["external_source_version", "text"], aggregationProvenance: [ "aggregation_provenance", "enum:measurement_aggregation_provenance", ], glucoseContext: ["glucose_context", "enum:glucose_context"], sleepStage: ["sleep_stage", "enum:sleep_stage"], rhythmClassification: ["rhythm_classification", "enum:rhythm_classification"], deviceType: ["device_type", "text"], ingestTokenId: ["ingest_token_id", "text"], syncVersion: ["sync_version", "int4"], deletedAt: ["deleted_at", "timestamp"], createdAt: ["created_at", "timestamp"], updatedAt: ["updated_at", "timestamp"], }; const FIELDS = Object.keys(COLUMNS) as Array; /** How each array travels, and how its element is turned back into a value. */ function arrayType(kind: Kind): string { if (kind === "float8" || kind === "int4") return `${kind}[]`; return "text[]"; // timestamps as ISO text, bytes as hex, enums as labels } function selectExpression( field: keyof MeasurementInsertRow, kind: Kind, ): string { const ref = `u."${field}"`; if (kind === "timestamp") { // Stored as UTC in `timestamp(3)`. ISO text with a `Z` casts to the same // wall-clock instant; the zone suffix is ignored by the cast. const cast = `${ref}::timestamp(3)`; // Prisma fills both on create when the row does not carry them. return field === "createdAt" || field === "updatedAt" ? `COALESCE(${cast}, CURRENT_TIMESTAMP AT TIME ZONE 'UTC')` : cast; } if (kind === "bytea") return `decode(${ref}, 'hex')`; if (kind.startsWith("enum:")) return `${ref}::${kind.slice(5)}`; return ref; } /** * The statement, built once. Its text depends on nothing but `COLUMNS`, a * compile-time table: no input reaches it, and every value travels as a bound * array parameter. */ export const MEASUREMENT_BULK_INSERT_SQL = ` INSERT INTO measurements (${FIELDS.map((f) => COLUMNS[f][0]).join(", ")}) SELECT ${FIELDS.map((f) => selectExpression(f, COLUMNS[f][1])).join(", ")} FROM unnest(${FIELDS.map((f, i) => `$${i + 1}::${arrayType(COLUMNS[f][1])}`).join(", ")}) AS u(${FIELDS.map((f) => `"${f}"`).join(", ")}) `; function cell(value: unknown, kind: Kind): unknown { if (value === null || value === undefined) return null; if (kind === "timestamp") return (value as Date).toISOString(); if (kind === "bytea") return Buffer.from(value as Uint8Array).toString("hex"); return value; } /** * Rows one statement carries at most. See the module comment: the parameters * of roughly the last hundred statements stay in memory, so this, not the * caller's batch, is what bounds that memory. */ export const MEASUREMENT_INSERT_ROWS_PER_STATEMENT = 100; /** The rows as parameter arrays, one per column, for one statement. */ function columnArrays(rows: readonly MeasurementInsertRow[]): unknown[][] { return FIELDS.map((field) => { const kind = COLUMNS[field][1]; return rows.map((row) => cell(row[field], kind)); }); } /** * Insert `rows`, {@link MEASUREMENT_INSERT_ROWS_PER_STATEMENT} to a statement. * Resolves to the number inserted. */ export async function insertMeasurementRows( tx: Prisma.TransactionClient, rows: readonly MeasurementInsertRow[], ): Promise { let inserted = 0; for ( let at = 0; at < rows.length; at += MEASUREMENT_INSERT_ROWS_PER_STATEMENT ) { const slice = rows.slice(at, at + MEASUREMENT_INSERT_ROWS_PER_STATEMENT); inserted += await tx.$executeRawUnsafe( MEASUREMENT_BULK_INSERT_SQL, ...columnArrays(slice), ); } return inserted; } /** * The same statement, skipping any row that collides with an existing one on * either unique identity, and answering with the rows it did insert. * * `ON CONFLICT DO NOTHING` with no target is what Prisma emits for * `createMany({ skipDuplicates: true })`: a row that meets either * `(user_id, type, source, external_id)` or * `(user_id, type, measured_at, source, sleep_stage)` is left out, a row that * meets another row of the same statement is left out after the first, and * nothing throws for either. */ export const MEASUREMENT_BULK_INSERT_SKIP_DUPLICATES_SQL = `${MEASUREMENT_BULK_INSERT_SQL} ON CONFLICT DO NOTHING RETURNING id, type::text AS "type", measured_at AS "measuredAt", external_id AS "externalId" `; /** One row `insertNewMeasurementRows` wrote. */ export interface InsertedMeasurementRow { id: string; type: MeasurementType; measuredAt: Date; externalId: string | null; } /** A row for {@link insertNewMeasurementRows}: the id and the defaults are filled in. */ export type NewMeasurementRow = Omit< MeasurementInsertRow, | "id" | "valueMin" | "valueMax" | "notes" | "notesEncrypted" | "aggregationProvenance" | "glucoseContext" | "rhythmClassification" | "syncVersion" | "deletedAt" | "createdAt" | "updatedAt" > & Partial< Pick< MeasurementInsertRow, | "valueMin" | "valueMax" | "notesEncrypted" | "aggregationProvenance" | "glucoseContext" | "rhythmClassification" > >; /** * Insert brand-new rows, skipping duplicates, and return the * rows that landed. The replacement for * `measurement.createManyAndReturn({ data, skipDuplicates: true })`. * * Every column the schema defaults is written with that default: a fresh * cuid-shaped id, `sync_version` 1, not deleted, created and updated at the * moment of the call, and null for every optional column the row does not * carry. */ export async function insertNewMeasurementRows( db: Pick | Prisma.TransactionClient, rows: readonly NewMeasurementRow[], ): Promise { if (rows.length === 0) return []; // Stamped here, as the Prisma client stamps `@default(now())` and // `@updatedAt`, rather than left to the database clock. const now = new Date(); const full: MeasurementInsertRow[] = rows.map((row) => ({ id: newMeasurementId(), userId: row.userId, type: row.type, value: row.value, valueMin: row.valueMin ?? null, valueMax: row.valueMax ?? null, unit: row.unit, source: row.source, measuredAt: row.measuredAt, notes: null, notesEncrypted: row.notesEncrypted ?? null, externalId: row.externalId, externalSourceVersion: row.externalSourceVersion, aggregationProvenance: row.aggregationProvenance ?? null, glucoseContext: row.glucoseContext ?? null, sleepStage: row.sleepStage, rhythmClassification: row.rhythmClassification ?? null, deviceType: row.deviceType, syncVersion: 1, deletedAt: null, createdAt: now, updatedAt: now, })); const inserted: InsertedMeasurementRow[] = []; for ( let at = 0; at < full.length; at += MEASUREMENT_INSERT_ROWS_PER_STATEMENT ) { const slice = full.slice(at, at + MEASUREMENT_INSERT_ROWS_PER_STATEMENT); inserted.push( ...(await db.$queryRawUnsafe( MEASUREMENT_BULK_INSERT_SKIP_DUPLICATES_SQL, ...columnArrays(slice), )), ); } return inserted; } // ── Ids ────────────────────────────────────────────────────────────────── // // Prisma mints `@default(cuid())` in the client, so a row written around the // client needs one minted here. Same shape as Prisma's own (cuid v1): `c`, the // time in base 36, a rolling counter, a host fingerprint, and eight random // characters, 25 characters in all. Ids stay roughly time-ordered, which the // list reads rely on only as a tiebreaker. const BLOCK = 4; const BLOCK_SPACE = 36 ** BLOCK; let counter = randomInt(BLOCK_SPACE); function block(n: number): string { return n.toString(36).padStart(BLOCK, "0").slice(-BLOCK); } const FINGERPRINT = (() => { const host = hostname(); let sum = host.length + 36; for (let i = 0; i < host.length; i++) sum += host.charCodeAt(i); return ( (process.pid % 1296).toString(36).padStart(2, "0") + (sum % 1296).toString(36).padStart(2, "0") ); })(); /** A fresh id in the shape Prisma's `cuid()` default produces. */ export function newMeasurementId(): string { counter = (counter + 1) % BLOCK_SPACE; return ( "c" + Date.now().toString(36).padStart(8, "0") + block(counter) + FINGERPRINT + block(randomInt(BLOCK_SPACE)) + block(randomInt(BLOCK_SPACE)) ); }