Bulk import with 1000 conversations: cost and performance checks

Bulk import pays for source discovery, reading and parsing, normalization, SQL writes, and database export. Summarization, embeddings, and relation discovery form a separate pipeline. Measure the stages separately before changing object retention, export frequency, or model requests. Here, 1000 is a synthetic fixture size; the original personal timings, OOM story, and claimed speedups are not verified results.

What the pinned implementation already does

server/src/services/import.ts runs detect/scan sequentially for enabled adapters, collects their metadata into allMetas, then checks, parses, and writes each conversation. Matching size and modification time can skip parsing; changed metadata with an unchanged content hash updates metadata only. Recording parserVersion does not force a reparse, as explained in adapter internals.

Each parsed conversation still contains its full messages array. Writes use a per-conversation transaction and a prepared message statement. Replacement invalidates old automatic notes and schedules vector cleanup while preserving eligible user-edited notes or rejection state. The batch calls saveDatabase at the end. Automatic saving in server/src/db/index.ts remains enabled by server startup; importAll does not temporarily disable it. Per-item export, automatic save suspension, and a streaming parseAndInsert API are not current behavior.

Import returns imported, replaced, skipped, errors, their IDs, and summarizationCandidateIds. Candidate IDs are not completed summaries. server/src/services/summarize.ts separately runs experience filtering, summarization, embeddings, and relation discovery. A conversation may become filtered; an embedding failure need not fail its summary.

Define costs that can be checked

Stage	Record together	Check first
detect/scan	Source, files, metadata count, duration	Discovered files match expected sources
parse/hash	Messages, bytes, largest conversation	Content fields and order stay intact
SQL/transaction	Imported, replaced, skipped, failed IDs	A failed conversation leaves no partial record
export/save	Bytes, call count, reopened data	Separate memory commit from file persistence
Summary/embedding/relations	Queued and final states, requests, errors	Receipt is not completion or quality

Node’s memoryUsage documentation distinguishes RSS, V8 heap, and external memory; arrayBuffers is already included in external, so do not add it twice. Sampling can miss a short synchronous export spike, and an end snapshot is not a run-wide peak. Record sampling, runtime and dependency versions, hardware, fixtures, and cache conditions. Conversation count alone cannot establish a fixed memory threshold.

Compare retention with 1000 synthetic inputs

In a separate directory, save this as performance.mjs under Node 24. Run npm install --ignore-scripts --save-exact sql.js@1.14.2, then node performance.mjs. It creates temporary normalized JSONL and isolated WASM databases. One path retains all input objects; the other inserts lines through readline. Both use per-conversation transactions, per-conversation prepared message statements, and one final export with a reopen check.

The exercise compares teaching-input retention and final data. It does not run ChatCrystal discovery, vendor parsers, automatic saving, or models, and does not establish streaming insertion in current adapters. The 1000 and 1 counters describe explicitly retained normalized objects, not buffer size, whole-database memory, or process peak. Timings and end snapshots depend on the machine and execution order.

import assert from 'node:assert/strict';
import { mkdtemp, writeFile, readFile, rm } from 'node:fs/promises';
import { createReadStream } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { createInterface } from 'node:readline';
import { performance } from 'node:perf_hooks';
import initSqlJs from 'sql.js';

const SQL = await initSqlJs();
const dir = await mkdtemp(join(tmpdir(), 'import-fixture-'));
const file = join(dir, 'normalized.jsonl');
const count = 1000;
const rows = Array.from({ length: count }, (_, i) => ({
  id: `fixture-${i}`,
  messages: [
    { role: 'user', content: `question ${i}` },
    { role: 'assistant', content: `answer ${i}` },
  ],
}));
await writeFile(file, rows.map(JSON.stringify).join('\n') + '\n');
rows.length = 0;

async function run(retainAll) {
  const db = new SQL.Database();
  db.run(
    'CREATE TABLE messages (conversation TEXT, position INTEGER, role TEXT, content TEXT, PRIMARY KEY (conversation, position))'
  );
  let peakRetained = 0;
  let imported = 0;
  const started = performance.now();
  const insert = (item) => {
    assert.equal(item.messages.length, 2);
    db.run('BEGIN');
    const stmt = db.prepare('INSERT INTO messages VALUES (?, ?, ?, ?)');
    try {
      item.messages.forEach((m, i) => stmt.run([item.id, i, m.role, m.content]));
      db.run('COMMIT');
      imported++;
    } catch (error) {
      db.run('ROLLBACK');
      throw error;
    } finally {
      stmt.free();
    }
  };
  try {
    if (retainAll) {
      const items = (await readFile(file, 'utf8')).trim().split('\n').map(JSON.parse);
      peakRetained = items.length;
      items.forEach(insert);
    } else {
      const lines = createInterface({ input: createReadStream(file), crlfDelay: Infinity });
      try {
        for await (const line of lines) {
          if (!line.trim()) continue;
          peakRetained = Math.max(peakRetained, 1);
          insert(JSON.parse(line));
        }
      } finally {
        lines.close();
      }
    }
    const values = db.exec(
      'SELECT conversation, position, role, content FROM messages ORDER BY conversation, position'
    )[0].values;
    assert.equal(imported, count);
    assert.equal(values.length, count * 2);
    const bytes = db.export();
    const reopened = new SQL.Database(bytes);
    try {
      assert.equal(reopened.exec('SELECT count(*) FROM messages')[0].values[0][0], count * 2);
    } finally {
      reopened.close();
    }
    return {
      values,
      peakRetained,
      elapsedMs: performance.now() - started,
      exportedBytes: bytes.length,
      memory: process.memoryUsage(),
    };
  } finally {
    db.close();
  }
}

try {
  const retained = await run(true);
  const streamed = await run(false);
  assert.deepEqual(streamed.values, retained.values);
  assert.equal(retained.peakRetained, count);
  assert.equal(streamed.peakRetained, 1);
  const { values: a, ...first } = retained;
  const { values: b, ...second } = streamed;
  console.log(
    JSON.stringify(
      {
        scope: 'synthetic normalized JSONL; not ChatCrystal importAll',
        retained: first,
        streamed: second,
      },
      null,
      2
    )
  );
} finally {
  await rm(dir, { recursive: true, force: true });
}

Make acceptance fail before optimizing

Set the streamed peakRetained to count: the retention assertion must fail. Insert only user messages: count and equivalence checks must fail. Restore the original code and rerun. elapsedMs and final memory describe this observation only. The paths run sequentially in one process, so this is not evidence of a speed or RSS winner.

For real import, add malformed JSON, a very large conversation, duplicate or changed content, unwritable output, and persisted-file reopening. Streaming may reduce some input objects; it does not turn the WASM database into an out-of-core database. Preserve transactions, content hashes, replacement behavior, and recovery before adding an asynchronous streaming interface. See WASM persistence boundaries.

Check the summary queue and watcher separately

server/src/queue/index.ts uses concurrency 1, intervalCap 1, and interval 1000ms. These govern task starts, not a shared rate limit for every Provider HTTP request. SDK retries and task retries may add requests. The outer loop backs off only when an error string contains 429 and another attempt remains; its default allows up to four task attempts. Queue count cannot establish actual request count or completion time.

server/src/watcher/index.ts adds a three-second debounce after file-stability waiting. An import already running in that watcher sets pending and schedules another run after completion. That flag belongs to the watcher, not a global lock shared by every HTTP or CLI import. watcherPolicy disables watching in cloud mode. Fix the entry point and observe task states rather than collapsing file events, parse runs, and model requests into one throughput number.