perf(import): optimize write and FTS batches

apps/desktop/preload/index.d.ts

@@ -58,6 +58,7 @@ interface ImportDiagnostics {
     }
     messageBatchCount: number
     messageTransactionCount: number
+    messageInsertStatementCount: number
     rssStartMb: number
     /** Highest RSS observed at import stage and parser batch boundaries; not a continuous process peak. */
     rssSampledPeakMb: number

packages/node-runtime/src/fts/index.test.ts

@@ -0,0 +1,46 @@
+import assert from 'node:assert/strict'
+import path from 'node:path'
+import test from 'node:test'
+import Database from 'better-sqlite3'
+import { BetterSqliteAdapter } from '../better-sqlite3-adapter'
+import { buildFtsIndex, searchByFts } from './index'
+
+const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
+
+test('buildFtsIndex indexes sparse eligible rows exactly once across keyset batches', () => {
+  const raw = new Database(':memory:', { nativeBinding })
+  raw.exec(`
+    CREATE TABLE message (
+      id INTEGER PRIMARY KEY,
+      type INTEGER NOT NULL,
+      content TEXT
+    )
+  `)
+  const insert = raw.prepare('INSERT INTO message (id, type, content) VALUES (?, ?, ?)')
+  const expectedRowIds = [-2, 0]
+  insert.run(-2, 0, 'negative marker')
+  insert.run(0, 0, 'zero marker')
+
+  for (let index = 1; index <= 5_001; index++) {
+    const id = index * 2
+    insert.run(id, 0, `marker${index}`)
+    expectedRowIds.push(id)
+  }
+  insert.run(10_003, 1, 'non text')
+  insert.run(10_004, 0, '')
+  insert.run(10_005, 0, null)
+  insert.run(10_006, 0, '   ')
+
+  const db = new BetterSqliteAdapter(raw)
+  const result = buildFtsIndex(db)
+  const actualRowIds = raw
+    .prepare('SELECT rowid FROM message_fts ORDER BY rowid')
+    .all()
+    .map((row) => (row as { rowid: number }).rowid)
+
+  assert.equal(result.indexed, expectedRowIds.length + 1)
+  assert.deepEqual(actualRowIds, expectedRowIds)
+  assert.deepEqual(searchByFts(db, ['marker1']).rowids, [2])
+  assert.deepEqual(searchByFts(db, ['marker5001']).rowids, [10_002])
+  raw.close()
+})

packages/node-runtime/src/fts/index.ts

@@ -39,23 +39,30 @@ export function buildFtsIndex(db: DatabaseAdapter): { indexed: number } {
 
   const insertFts = db.prepare('INSERT INTO message_fts(rowid, content) VALUES (?, ?)')
 
-  const countRow = db
-    .prepare("SELECT COUNT(*) as total FROM message WHERE type = 0 AND content IS NOT NULL AND content != ''")
-    .get() as { total: number } | undefined
-  const total = countRow?.total ?? 0
-
   let indexed = 0
-  let offset = 0
-
-  while (offset < total) {
-    const rows = db
-      .prepare(
-        `SELECT id, content FROM message
-         WHERE type = 0 AND content IS NOT NULL AND content != ''
-         ORDER BY id ASC LIMIT ? OFFSET ?`
-      )
-      .all(BATCH_SIZE, offset) as Array<{ id: number; content: string }>
+  let lastId: number | null = null
+  const selectFirstBatch = db.prepare(
+    `SELECT id, content FROM message
+     WHERE type = 0
+       AND content IS NOT NULL
+       AND content != ''
+     ORDER BY id ASC
+     LIMIT ?`
+  )
+  const selectNextBatch = db.prepare(
+    `SELECT id, content FROM message
+     WHERE type = 0
+       AND content IS NOT NULL
+       AND content != ''
+       AND id > ?
+     ORDER BY id ASC
+     LIMIT ?`
+  )
 
+  while (true) {
+    const rows = (
+      lastId === null ? selectFirstBatch.all(BATCH_SIZE) : selectNextBatch.all(lastId, BATCH_SIZE)
+    ) as Array<{ id: number; content: string }>
     if (rows.length === 0) break
 
     db.transaction(() => {
@@ -68,7 +75,7 @@ export function buildFtsIndex(db: DatabaseAdapter): { indexed: number } {
     })
 
     indexed += rows.length
-    offset += BATCH_SIZE
+    lastId = rows[rows.length - 1].id
   }
 
   return { indexed }

packages/node-runtime/src/import/auto-importer.test.ts

@@ -77,6 +77,7 @@ function createDeps(options?: {
             },
             messageBatchCount: 0,
             messageTransactionCount: 0,
+            messageInsertStatementCount: 0,
             rssStartMb: 0,
             rssSampledPeakMb: 0,
             rssSampledDeltaMb: 0,

packages/node-runtime/src/import/index.ts

@@ -14,6 +14,7 @@ export {
   logSummary,
 } from './perf-logger'
 export { streamingImport, analyzeNewImport, streamParseFileInfo } from './streaming-importer'
+export { MessageBatchInserter, MESSAGE_INSERT_MAX_ROWS } from './message-batch-inserter'
 export type {
   SkipReasons,
   ImportDiagnostics,
@@ -28,6 +29,7 @@ export type {
   StreamParseFileInfoResult,
   StreamParseFileInfoDeps,
 } from './streaming-importer'
+export type { MessageInsertRow } from './message-batch-inserter'
 export { analyzeIncrementalImport, incrementalImport } from './incremental-importer'
 export type {
   ImportOptions,

packages/node-runtime/src/import/message-batch-inserter.test.ts

@@ -0,0 +1,107 @@
+import assert from 'node:assert/strict'
+import path from 'node:path'
+import test from 'node:test'
+import Database from 'better-sqlite3'
+import { BetterSqliteAdapter } from '../better-sqlite3-adapter'
+import { MESSAGE_INSERT_MAX_ROWS, MessageBatchInserter, type MessageInsertRow } from './message-batch-inserter'
+
+const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
+const messageSchema = `
+  CREATE TABLE message (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    sender_id INTEGER NOT NULL,
+    sender_account_name TEXT,
+    sender_group_nickname TEXT,
+    ts INTEGER NOT NULL,
+    type INTEGER NOT NULL CHECK(type >= 0),
+    content TEXT,
+    reply_to_message_id TEXT,
+    platform_message_id TEXT
+  )
+`
+
+function makeRows(count: number): MessageInsertRow[] {
+  return Array.from({ length: count }, (_, index) => ({
+    senderId: (index % 7) + 1,
+    senderAccountName: index % 3 === 0 ? null : `account-${index % 7}`,
+    senderGroupNickname: index % 4 === 0 ? '' : `nickname-${index % 5}`,
+    timestamp: 1_700_000_000 + index,
+    type: index % 10,
+    content: index % 6 === 0 ? null : `content-${index}`,
+    replyToMessageId: index % 8 === 0 ? `reply-${index - 1}` : null,
+    platformMessageId: index % 9 === 0 ? null : `message-${index}`,
+  }))
+}
+
+function readMessages(db: Database.Database): Record<string, unknown>[] {
+  return db
+    .prepare(
+      `SELECT
+         id,
+         sender_id,
+         sender_account_name,
+         sender_group_nickname,
+         ts,
+         type,
+         content,
+         reply_to_message_id,
+         platform_message_id
+       FROM message
+       ORDER BY id`
+    )
+    .all() as Record<string, unknown>[]
+}
+
+test('MessageBatchInserter preserves row order and nullable values across SQLite variable-limit chunks', () => {
+  const batchRaw = new Database(':memory:', { nativeBinding })
+  const referenceRaw = new Database(':memory:', { nativeBinding })
+  batchRaw.exec(messageSchema)
+  referenceRaw.exec(messageSchema)
+
+  const rows = makeRows(MESSAGE_INSERT_MAX_ROWS * 2 + 7)
+  const statementCount = new MessageBatchInserter(new BetterSqliteAdapter(batchRaw)).insert(rows)
+  const referenceInsert = referenceRaw.prepare(
+    `INSERT INTO message (
+       sender_id,
+       sender_account_name,
+       sender_group_nickname,
+       ts,
+       type,
+       content,
+       reply_to_message_id,
+       platform_message_id
+     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+  )
+  for (const row of rows) {
+    referenceInsert.run(
+      row.senderId,
+      row.senderAccountName,
+      row.senderGroupNickname,
+      row.timestamp,
+      row.type,
+      row.content,
+      row.replyToMessageId,
+      row.platformMessageId
+    )
+  }
+
+  assert.equal(statementCount, 3)
+  assert.deepEqual(readMessages(batchRaw), readMessages(referenceRaw))
+  batchRaw.close()
+  referenceRaw.close()
+})
+
+test('MessageBatchInserter failure can roll back the complete active transaction', () => {
+  const raw = new Database(':memory:', { nativeBinding })
+  raw.exec(messageSchema)
+  const db = new BetterSqliteAdapter(raw)
+  const rows = makeRows(3)
+  rows[1] = { ...rows[1], type: -1 }
+
+  db.exec('BEGIN TRANSACTION')
+  assert.throws(() => new MessageBatchInserter(db).insert(rows), /CHECK constraint failed/)
+  db.exec('ROLLBACK')
+
+  assert.equal((raw.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number }).count, 0)
+  raw.close()
+})

packages/node-runtime/src/import/message-batch-inserter.ts

@@ -0,0 +1,73 @@
+import type { DatabaseAdapter, PreparedStatement } from '@openchatlab/core'
+
+const MESSAGE_COLUMN_COUNT = 8
+const SQLITE_LEGACY_VARIABLE_LIMIT = 999
+
+/**
+ * Keep each statement below SQLite's historical 999-variable limit.
+ * Modern bundled SQLite builds allow more, but this conservative bound keeps
+ * the shared DatabaseAdapter path compatible with older/runtime-specific builds.
+ */
+export const MESSAGE_INSERT_MAX_ROWS = Math.floor(SQLITE_LEGACY_VARIABLE_LIMIT / MESSAGE_COLUMN_COUNT)
+
+export interface MessageInsertRow {
+  senderId: number
+  senderAccountName: string | null
+  senderGroupNickname: string | null
+  timestamp: number
+  type: number
+  content: string | null
+  replyToMessageId: string | null
+  platformMessageId: string | null
+}
+
+export class MessageBatchInserter {
+  private readonly statementCache = new Map<number, PreparedStatement>()
+
+  constructor(private readonly db: DatabaseAdapter) {}
+
+  insert(rows: readonly MessageInsertRow[]): number {
+    let statementCount = 0
+    for (let offset = 0; offset < rows.length; offset += MESSAGE_INSERT_MAX_ROWS) {
+      const batch = rows.slice(offset, offset + MESSAGE_INSERT_MAX_ROWS)
+      const statement = this.getStatement(batch.length)
+      statement.run(...batch.flatMap(toParams))
+      statementCount++
+    }
+    return statementCount
+  }
+
+  private getStatement(rowCount: number): PreparedStatement {
+    const cached = this.statementCache.get(rowCount)
+    if (cached) return cached
+
+    const values = Array.from({ length: rowCount }, () => '(?, ?, ?, ?, ?, ?, ?, ?)').join(', ')
+    const statement = this.db.prepare(
+      `INSERT INTO message (
+         sender_id,
+         sender_account_name,
+         sender_group_nickname,
+         ts,
+         type,
+         content,
+         reply_to_message_id,
+         platform_message_id
+       ) VALUES ${values}`
+    )
+    this.statementCache.set(rowCount, statement)
+    return statement
+  }
+}
+
+function toParams(row: MessageInsertRow): unknown[] {
+  return [
+    row.senderId,
+    row.senderAccountName,
+    row.senderGroupNickname,
+    row.timestamp,
+    row.type,
+    row.content,
+    row.replyToMessageId,
+    row.platformMessageId,
+  ]
+}

packages/node-runtime/src/import/streaming-importer.test.ts

@@ -112,8 +112,9 @@ function createCollectingLogger(messages: string[]): ImportLogger {
 }
 
 function assertPerformanceDiagnostics(result: StreamImportResult): void {
-  const diagnostics = result.diagnostics?.performance
-  assert.ok(diagnostics)
+  const importDiagnostics = result.diagnostics
+  assert.ok(importDiagnostics)
+  const diagnostics = importDiagnostics.performance
 
   for (const [stage, durationMs] of Object.entries(diagnostics.timings)) {
     assert.equal(Number.isFinite(durationMs), true, `${stage} should be finite`)
@@ -124,6 +125,8 @@ function assertPerformanceDiagnostics(result: StreamImportResult): void {
   assert.ok(diagnostics.timings.totalMs >= diagnostics.timings.messageWriteMs)
   assert.ok(diagnostics.messageBatchCount > 0)
   assert.ok(diagnostics.messageTransactionCount > 0)
+  assert.ok(diagnostics.messageInsertStatementCount > 0)
+  assert.ok(diagnostics.messageInsertStatementCount < importDiagnostics.messagesWritten)
   assert.ok(diagnostics.rssSampledPeakMb >= diagnostics.rssStartMb)
   assert.ok(diagnostics.rssSampledDeltaMb >= 0)
 }

packages/node-runtime/src/import/streaming-importer.ts

@@ -27,6 +27,7 @@ import {
 import * as fs from 'fs'
 import { performance } from 'node:perf_hooks'
 import { buildFtsIndex } from '../fts'
+import { MessageBatchInserter, type MessageInsertRow } from './message-batch-inserter'
 import { createMessageDedupState, registerMessageAndCheckDuplicate, type DedupMessage } from './message-deduplicator'
 
 // ==================== Public interfaces ====================
@@ -70,6 +71,7 @@ export interface ImportPerformanceDiagnostics {
   timings: ImportStageTimings
   messageBatchCount: number
   messageTransactionCount: number
+  messageInsertStatementCount: number
   rssStartMb: number
   /** Highest RSS observed at import stage and parser batch boundaries; not a continuous process peak. */
   rssSampledPeakMb: number
@@ -362,10 +364,6 @@ async function streamImportSingle(
              roles = CASE WHEN excluded.roles != '[]' THEN excluded.roles ELSE roles END`
         ),
         getMemberId: db.prepare(`SELECT id FROM member WHERE platform_id = ?`),
-        insertMessage: db.prepare(
-          `INSERT INTO message (sender_id, sender_account_name, sender_group_nickname, ts, type, content, reply_to_message_id, platform_message_id)
-           VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
-        ),
         insertNameHistory: db.prepare(
           `INSERT INTO member_name_history (member_id, name_type, name, start_ts, end_ts) VALUES (?, ?, ?, ?, ?)`
         ),
@@ -397,7 +395,6 @@ async function streamImportSingle(
     insertMeta,
     insertMember,
     getMemberId,
-    insertMessage,
     insertNameHistory,
     updateMemberAccountName,
     updateMemberGroupNickname,
@@ -422,6 +419,8 @@ async function streamImportSingle(
   let inTransaction = false
   let messageTransactionCount = 0
   let messageTransactionMs = 0
+  let messageInsertStatementCount = 0
+  const messageBatchInserter = new MessageBatchInserter(db)
 
   const beginTransaction = () => {
     if (!inTransaction) {
@@ -596,6 +595,13 @@ async function streamImportSingle(
           const transactionMsBefore = messageTransactionMs
           const checkpointMsBefore = timings.checkpointMs
           try {
+            const pendingRows: MessageInsertRow[] = []
+            const flushPendingRows = () => {
+              if (pendingRows.length === 0) return
+              messageInsertStatementCount += messageBatchInserter.insert(pendingRows)
+              pendingRows.length = 0
+            }
+
             callbackStats.onMessageBatchCalls++
             callbackStats.totalMessagesReceived += messages.length
             if (callbackStats.onMessageBatchCalls <= 3 || callbackStats.onMessageBatchCalls % 10 === 0) {
@@ -642,16 +648,16 @@ async function streamImportSingle(
               const senderId = memberIdMap.get(msg.senderPlatformId)
               if (senderId === undefined) continue
 
-              insertMessage.run(
+              pendingRows.push({
                 senderId,
-                senderAccountName || null,
-                senderGroupNickname || null,
-                dedupMessage.timestamp,
-                dedupMessage.type,
-                dedupMessage.content,
-                msg.replyToMessageId || null,
-                msg.platformMessageId || null
-              )
+                senderAccountName: senderAccountName || null,
+                senderGroupNickname: senderGroupNickname || null,
+                timestamp: dedupMessage.timestamp,
+                type: dedupMessage.type,
+                content: dedupMessage.content,
+                replyToMessageId: msg.replyToMessageId || null,
+                platformMessageId: msg.platformMessageId || null,
+              })
               messageCountInBatch++
               totalMessageCount++
 
@@ -659,10 +665,12 @@ async function streamImportSingle(
               trackNickname(groupNicknameTracker, msg.senderPlatformId, senderGroupNickname, msg.timestamp)
 
               if (messageCountInBatch >= BATCH_COMMIT_SIZE) {
+                flushPendingRows()
                 commitAndBeginNew()
                 messageCountInBatch = 0
               }
             }
+            flushPendingRows()
           } finally {
             const nestedStageMs =
               messageTransactionMs - transactionMsBefore + (timings.checkpointMs - checkpointMsBefore)
@@ -872,6 +880,7 @@ async function streamImportSingle(
       timings,
       messageBatchCount: callbackStats.onMessageBatchCalls,
       messageTransactionCount,
+      messageInsertStatementCount,
       rssStartMb: rssStartBytes / bytesPerMegabyte,
       rssSampledPeakMb: rssPeakBytes / bytesPerMegabyte,
       rssSampledDeltaMb: (rssPeakBytes - rssStartBytes) / bytesPerMegabyte,

packages/node-runtime/src/index.ts

@@ -38,6 +38,8 @@ export {
   ArchiveImportError,
   GoogleChatTakeoutResolver,
   ArchiveImportSourceManager,
+  MessageBatchInserter,
+  MESSAGE_INSERT_MAX_ROWS,
 } from './import'
 export type {
   ImportMeta,
@@ -74,6 +76,7 @@ export type {
   PreparedImportSource,
   MaterializedImport,
   ArchiveResolver,
+  MessageInsertRow,
 } from './import'
 export {
   LogLevel,

src/services/import/types.ts

@@ -87,6 +87,7 @@ export interface ImportPerformanceDiagnosticsInfo {
   }
   messageBatchCount: number
   messageTransactionCount: number
+  messageInsertStatementCount: number
   rssStartMb: number
   /** Highest RSS observed at import stage and parser batch boundaries; not a continuous process peak. */
   rssSampledPeakMb: number