diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index c3df4b63..1cfc6d78 100644 --- a/apps/desktop/preload/index.d.ts +++ b/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 diff --git a/packages/node-runtime/src/fts/index.test.ts b/packages/node-runtime/src/fts/index.test.ts new file mode 100644 index 00000000..46ca7b0d --- /dev/null +++ b/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() +}) diff --git a/packages/node-runtime/src/fts/index.ts b/packages/node-runtime/src/fts/index.ts index 64984c78..781b8c40 100644 --- a/packages/node-runtime/src/fts/index.ts +++ b/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 } diff --git a/packages/node-runtime/src/import/auto-importer.test.ts b/packages/node-runtime/src/import/auto-importer.test.ts index 98e7c928..c80db8df 100644 --- a/packages/node-runtime/src/import/auto-importer.test.ts +++ b/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, diff --git a/packages/node-runtime/src/import/index.ts b/packages/node-runtime/src/import/index.ts index c8feb5b4..03d3c466 100644 --- a/packages/node-runtime/src/import/index.ts +++ b/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, diff --git a/packages/node-runtime/src/import/message-batch-inserter.test.ts b/packages/node-runtime/src/import/message-batch-inserter.test.ts new file mode 100644 index 00000000..43b10d69 --- /dev/null +++ b/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[] { + 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[] +} + +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() +}) diff --git a/packages/node-runtime/src/import/message-batch-inserter.ts b/packages/node-runtime/src/import/message-batch-inserter.ts new file mode 100644 index 00000000..642a54c9 --- /dev/null +++ b/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() + + 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, + ] +} diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index 92b76e1e..51eee213 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/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) } diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index 03c41ed4..ea0b3484 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/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, diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 98211ce4..6a860b6c 100644 --- a/packages/node-runtime/src/index.ts +++ b/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, diff --git a/src/services/import/types.ts b/src/services/import/types.ts index 007e759c..002ff278 100644 --- a/src/services/import/types.ts +++ b/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