diff --git a/apps/cli/src/import/chatlab-reader.ts b/apps/cli/src/import/chatlab-reader.ts deleted file mode 100644 index a58ee330..00000000 --- a/apps/cli/src/import/chatlab-reader.ts +++ /dev/null @@ -1,233 +0,0 @@ -/** - * ChatLab 格式读取器 - * - * 支持 ChatLab JSON (.json) 和 ChatLab JSONL (.jsonl) 两种格式。 - * 不依赖 stream-json，使用原生 readline 和 JSON.parse 实现。 - */ - -import * as fs from 'fs' -import * as readline from 'readline' - -export interface ImportMeta { - name: string - platform: string - type: string - groupId?: string - groupAvatar?: string - ownerId?: string -} - -export interface ImportMember { - platformId: string - accountName: string - groupNickname?: string - avatar?: string - roles?: Array<{ id: string; name?: string }> -} - -export interface ImportMessage { - platformMessageId?: string - senderPlatformId: string - senderAccountName: string - senderGroupNickname?: string - timestamp: number - type: number - content: string | null - replyToMessageId?: string -} - -export interface ParsedData { - meta: ImportMeta - members: ImportMember[] - messages: ImportMessage[] -} - -export type ProgressCallback = (processed: number, total: number) => void - -/** - * 检测文件是否为 ChatLab 格式 - */ -export function detectFormat(filePath: string): 'chatlab-json' | 'chatlab-jsonl' | null { - const ext = filePath.toLowerCase() - if (ext.endsWith('.jsonl')) { - const head = fs.readFileSync(filePath, { encoding: 'utf-8', flag: 'r' }).slice(0, 1024) - if (head.includes('"_type"') && head.includes('"chatlab"')) { - return 'chatlab-jsonl' - } - return null - } - if (ext.endsWith('.json')) { - const head = fs.readFileSync(filePath, { encoding: 'utf-8', flag: 'r' }).slice(0, 4096) - if (/"chatlab"\s*:\s*\{/.test(head)) { - return 'chatlab-json' - } - return null - } - return null -} - -/** - * 解析 ChatLab JSON 格式文件 - */ -export async function parseChatLabJson(filePath: string, onProgress?: ProgressCallback): Promise { - const raw = fs.readFileSync(filePath, 'utf-8') - const data = JSON.parse(raw) - - if (!data.chatlab || typeof data.chatlab !== 'object') { - throw new Error('Invalid ChatLab JSON: missing "chatlab" field') - } - - const meta: ImportMeta = { - name: data.meta?.name || '未知', - platform: data.meta?.platform || 'unknown', - type: data.meta?.type || 'group', - groupId: data.meta?.groupId, - groupAvatar: data.meta?.groupAvatar, - ownerId: data.meta?.ownerId, - } - - const members: ImportMember[] = (data.members || []).map((m: Record) => ({ - platformId: String(m.platformId || ''), - accountName: String(m.accountName || m.platformId || ''), - groupNickname: m.groupNickname ? String(m.groupNickname) : undefined, - avatar: m.avatar ? String(m.avatar) : undefined, - roles: m.roles as ImportMember['roles'], - })) - - const rawMessages = data.messages || [] - const totalMessages = rawMessages.length - - const messages: ImportMessage[] = rawMessages.map((msg: Record, i: number) => { - if (onProgress && i % 10000 === 0) { - onProgress(i, totalMessages) - } - return { - platformMessageId: msg.platformMessageId ? String(msg.platformMessageId) : undefined, - senderPlatformId: String(msg.sender || ''), - senderAccountName: String(msg.accountName || msg.sender || ''), - senderGroupNickname: msg.groupNickname ? String(msg.groupNickname) : undefined, - timestamp: Number(msg.timestamp), - type: Number(msg.type ?? 0), - content: msg.content != null ? String(msg.content) : null, - replyToMessageId: msg.replyToMessageId ? String(msg.replyToMessageId) : undefined, - } - }) - - if (onProgress) onProgress(totalMessages, totalMessages) - - if (members.length === 0) { - collectMembersFromMessages(messages, members) - } - - return { meta, members, messages } -} - -/** - * 解析 ChatLab JSONL 格式文件 - */ -export async function parseChatLabJsonl(filePath: string, onProgress?: ProgressCallback): Promise { - const meta: ImportMeta = { name: '未知', platform: 'unknown', type: 'group' } - const members: ImportMember[] = [] - const messages: ImportMessage[] = [] - - const stat = fs.statSync(filePath) - let bytesRead = 0 - - const rl = readline.createInterface({ - input: fs.createReadStream(filePath, 'utf-8'), - crlfDelay: Infinity, - }) - - for await (const line of rl) { - bytesRead += Buffer.byteLength(line, 'utf-8') + 1 - - const trimmed = line.trim() - if (!trimmed) continue - - let obj: Record - try { - obj = JSON.parse(trimmed) - } catch { - continue - } - - const type = obj._type as string - - if (type === 'header' || (!type && obj.chatlab)) { - const m = obj.meta as Record | undefined - if (m) { - meta.name = String(m.name || '未知') - meta.platform = String(m.platform || 'unknown') - meta.type = String(m.type || 'group') - if (m.groupId) meta.groupId = String(m.groupId) - if (m.groupAvatar) meta.groupAvatar = String(m.groupAvatar) - if (m.ownerId) meta.ownerId = String(m.ownerId) - } - } else if (type === 'member') { - members.push({ - platformId: String(obj.platformId || ''), - accountName: String(obj.accountName || obj.platformId || ''), - groupNickname: obj.groupNickname ? String(obj.groupNickname) : undefined, - avatar: obj.avatar ? String(obj.avatar) : undefined, - roles: obj.roles as ImportMember['roles'], - }) - } else if (type === 'message') { - messages.push({ - platformMessageId: obj.platformMessageId ? String(obj.platformMessageId) : undefined, - senderPlatformId: String(obj.sender || obj.senderPlatformId || ''), - senderAccountName: String(obj.accountName || obj.senderAccountName || obj.sender || ''), - senderGroupNickname: obj.groupNickname - ? String(obj.groupNickname) - : obj.senderGroupNickname - ? String(obj.senderGroupNickname) - : undefined, - timestamp: Number(obj.timestamp), - type: Number(obj.type ?? 0), - content: obj.content != null ? String(obj.content) : null, - replyToMessageId: obj.replyToMessageId ? String(obj.replyToMessageId) : undefined, - }) - - if (onProgress && messages.length % 10000 === 0) { - onProgress(bytesRead, stat.size) - } - } - } - - if (onProgress) onProgress(stat.size, stat.size) - - if (members.length === 0) { - collectMembersFromMessages(messages, members) - } - - return { meta, members, messages } -} - -function collectMembersFromMessages(messages: ImportMessage[], members: ImportMember[]): void { - const seen = new Set() - for (const msg of messages) { - if (!seen.has(msg.senderPlatformId)) { - seen.add(msg.senderPlatformId) - members.push({ - platformId: msg.senderPlatformId, - accountName: msg.senderAccountName || msg.senderPlatformId, - groupNickname: msg.senderGroupNickname, - }) - } - } -} - -/** - * 统一入口：自动检测格式并解析 - */ -export async function parseFile(filePath: string, onProgress?: ProgressCallback): Promise { - const format = detectFormat(filePath) - if (!format) { - throw new Error(`Unsupported file format: ${filePath}. Only ChatLab JSON (.json) and JSONL (.jsonl) are supported.`) - } - - if (format === 'chatlab-json') { - return parseChatLabJson(filePath, onProgress) - } else { - return parseChatLabJsonl(filePath, onProgress) - } -} diff --git a/apps/cli/src/import/importer.test.ts b/apps/cli/src/import/importer.test.ts index c922f0b4..9f48ba00 100644 --- a/apps/cli/src/import/importer.test.ts +++ b/apps/cli/src/import/importer.test.ts @@ -5,7 +5,7 @@ import path from 'node:path' import test from 'node:test' import type { PathProvider } from '@openchatlab/core' import { DatabaseManager, raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta } from '@openchatlab/node-runtime' -import { importData } from './importer' +import { streamImport } from './stream-import' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -29,7 +29,23 @@ function createPathProvider(root: string): PathProvider { } } -test('importData raises the data directory gate after creating a current-schema database', async () => { +/** Write a minimal valid ChatLab Format JSON to a temp file and return the path. */ +function writeTempChatFile(dir: string): string { + const filePath = path.join(dir, 'test-chat.json') + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 1711468800 }, + meta: { name: 'Test Chat', platform: 'qq', type: 'group' }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + // accountName is required by streaming-importer (skipped otherwise) + messages: [{ sender: 'u1', accountName: 'Alice', timestamp: 1711468800, type: 0, content: 'hello' }], + }) + ) + return filePath +} + +test('streamImport raises the data directory gate after creating a current-schema database', async () => { const root = makeTempDir() fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) const manager = new DatabaseManager(createPathProvider(root), { @@ -37,23 +53,8 @@ test('importData raises the data directory gate after creating a current-schema runtime: { version: '0.25.1', kind: 'cli' }, }) - const result = await importData( - manager, - { - meta: { name: 'Fresh Import', platform: 'qq', type: 'group' }, - members: [{ platformId: 'u1', accountName: 'Alice' }], - messages: [ - { - senderPlatformId: 'u1', - senderAccountName: 'Alice', - timestamp: 1000, - type: 0, - content: 'hello', - }, - ], - }, - { nativeBinding } - ) + const chatFile = writeTempChatFile(root) + const result = await streamImport(manager, chatFile, { sessionId: 'test-session', nativeBinding }) assert.equal(result.success, true) const meta = readDataDirCompatibilityMeta(path.join(root, 'data')) @@ -62,7 +63,7 @@ test('importData raises the data directory gate after creating a current-schema assert.deepEqual(meta?.reasons, ['segment-schema']) }) -test('importData re-checks data directory compatibility before raw database writes', async () => { +test('streamImport re-checks data directory compatibility before raw database writes', async () => { const root = makeTempDir() fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) const pathProvider = createPathProvider(root) @@ -79,22 +80,11 @@ test('importData re-checks data directory compatibility before raw database writ runtime: { version: '0.25.1', kind: 'cli' }, }) - const result = await importData( - manager, - { - meta: { name: 'Blocked Import', platform: 'qq', type: 'group' }, - members: [{ platformId: 'u1', accountName: 'Alice' }], - messages: [ - { - senderPlatformId: 'u1', - senderAccountName: 'Alice', - timestamp: 1000, - type: 0, - content: 'hello', - }, - ], - }, - { nativeBinding } + const chatFile = writeTempChatFile(root) + // DataDirCompatibilityError is thrown before the inner try/catch in streaming-importer, + // so streamImport propagates it. Normalise to a result shape for assertions. + const result = await streamImport(manager, chatFile, { sessionId: 'test-session', nativeBinding }).catch( + (err: Error) => ({ success: false as const, error: err.message }) ) assert.equal(result.success, false) diff --git a/apps/cli/src/import/importer.ts b/apps/cli/src/import/importer.ts deleted file mode 100644 index 298308ac..00000000 --- a/apps/cli/src/import/importer.ts +++ /dev/null @@ -1,328 +0,0 @@ -/** - * ChatLab 数据导入器 - * - * 将解析后的数据写入 SQLite 数据库。 - * 支持新建导入和增量导入（基于 platform_message_id 或内容哈希去重）。 - */ - -import * as fs from 'fs' -import * as crypto from 'crypto' -import type { DatabaseAdapter } from '@openchatlab/core' -import { - CHAT_DB_SCHEMA, - FTS_TABLE_SCHEMA, - generateMessageKey, - buildMemberIdMap, - generateSessionIndex, - generateIncrementalSessionIndex, -} from '@openchatlab/core' -import type { DatabaseManager } from '@openchatlab/node-runtime' -import { writeParseResultToDb } from '@openchatlab/node-runtime' -import type { ParsedData, ImportMessage } from './chatlab-reader' - -export interface ImportResult { - success: boolean - sessionId: string - created: boolean - messageCount: number - memberCount: number - duplicateCount: number - error?: string -} - -export interface ImportOptions { - sessionId?: string - dryRun?: boolean - nativeBinding?: string -} - -function generateSessionId(): string { - const ts = Date.now() - const rand = crypto.randomBytes(4).toString('hex') - return `chat_${ts}_${rand}` -} - -/** - * 新建导入：创建新的会话数据库并导入全部数据 - * Delegates to shared writeParseResultToDb for data writing + name history. - */ -function fullImport( - db: DatabaseAdapter, - data: ParsedData, - onProgress?: (msg: string) => void -): { messageCount: number; memberCount: number; duplicateCount: number } { - db.exec(CHAT_DB_SCHEMA) - - onProgress?.(`写入 ${data.members.length} 个成员, ${data.messages.length} 条消息...`) - const stats = writeParseResultToDb(db, data.meta, data.members, data.messages) - - buildFts(db, onProgress) - - try { - generateSessionIndex(db) - } catch { - /* non-fatal */ - } - - return { messageCount: stats.messageCount, memberCount: stats.memberCount, duplicateCount: stats.skippedCount } -} - -/** - * 增量导入：向已有数据库追加新消息，自动去重 - */ -function incrementalImport( - db: DatabaseAdapter, - data: ParsedData, - onProgress?: (msg: string) => void -): { messageCount: number; memberCount: number; duplicateCount: number } { - onProgress?.('加载去重索引...') - const existingKeys = new Set() - - const existingPmids = db - .prepare('SELECT platform_message_id FROM message WHERE platform_message_id IS NOT NULL') - .all() as Array<{ platform_message_id: string }> - for (const row of existingPmids) { - existingKeys.add(`pmid:${row.platform_message_id}`) - } - - const existingHashes = db - .prepare( - `SELECT msg.ts, m.platform_id, msg.content - FROM message msg - JOIN member m ON msg.sender_id = m.id - WHERE msg.platform_message_id IS NULL` - ) - .all() as Array<{ ts: number; platform_id: string; content: string | null }> - for (const row of existingHashes) { - existingKeys.add(`hash:${generateMessageKey(row.ts, row.platform_id, row.content)}`) - } - - onProgress?.(`去重索引: ${existingKeys.size} 条已有消息`) - - const upsertMember = db.prepare( - `INSERT INTO member (platform_id, account_name, group_nickname, avatar, roles) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(platform_id) DO UPDATE SET - account_name = COALESCE(NULLIF(excluded.account_name, ''), account_name), - group_nickname = COALESCE(excluded.group_nickname, group_nickname)` - ) - - let newMembers = 0 - for (const m of data.members) { - const result = upsertMember.run( - m.platformId, - m.accountName || m.platformId, - m.groupNickname || null, - m.avatar || null, - m.roles ? JSON.stringify(m.roles) : '[]' - ) - if (result.changes > 0) newMembers++ - } - - const memberIdMap = buildMemberIdMap(db) - - const insertMsg = db.prepare( - `INSERT INTO message (sender_id, sender_account_name, sender_group_nickname, ts, type, content, reply_to_message_id, platform_message_id) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)` - ) - const insertMemberMinimal = db.prepare(`INSERT OR IGNORE INTO member (platform_id, account_name) VALUES (?, ?)`) - - let written = 0 - let duplicates = 0 - const BATCH = 5000 - - for (let i = 0; i < data.messages.length; i += BATCH) { - const batch = data.messages.slice(i, i + BATCH) - db.transaction(() => { - for (const msg of batch) { - if (isDuplicate(msg, existingKeys)) { - duplicates++ - continue - } - - let senderId = memberIdMap.get(msg.senderPlatformId) - if (!senderId) { - insertMemberMinimal.run(msg.senderPlatformId, msg.senderAccountName || msg.senderPlatformId) - senderId = ( - db.prepare('SELECT id FROM member WHERE platform_id = ?').get(msg.senderPlatformId) as { id: number } - )?.id - if (senderId) memberIdMap.set(msg.senderPlatformId, senderId) - } - if (!senderId) continue - - insertMsg.run( - senderId, - msg.senderAccountName || null, - msg.senderGroupNickname || null, - msg.timestamp, - msg.type, - msg.content, - msg.replyToMessageId || null, - msg.platformMessageId || null - ) - written++ - } - }) - - if (onProgress && (i + BATCH) % 50000 < BATCH) { - onProgress( - `已处理 ${Math.min(i + BATCH, data.messages.length)} / ${data.messages.length} 条消息 (新增 ${written}, 重复 ${duplicates})` - ) - } - } - - if (written > 0) { - insertFtsEntries(db, onProgress) - db.prepare('UPDATE meta SET imported_at = ?').run(Math.floor(Date.now() / 1000)) - try { - generateIncrementalSessionIndex(db) - } catch { - /* non-fatal */ - } - } - - return { messageCount: written, memberCount: newMembers, duplicateCount: duplicates } -} - -function isDuplicate(msg: ImportMessage, existingKeys: Set): boolean { - if (msg.platformMessageId) { - return existingKeys.has(`pmid:${msg.platformMessageId}`) - } - const key = generateMessageKey(msg.timestamp, msg.senderPlatformId, msg.content) - return existingKeys.has(`hash:${key}`) -} - -function buildFts(db: DatabaseAdapter, onProgress?: (msg: string) => void): void { - onProgress?.('创建全文搜索索引...') - db.exec(FTS_TABLE_SCHEMA) - - const textMessages = db - .prepare("SELECT id, content FROM message WHERE type = 0 AND content IS NOT NULL AND content != ''") - .all() as Array<{ id: number; content: string }> - - const insertFts = db.prepare('INSERT INTO message_fts(rowid, content) VALUES (?, ?)') - const BATCH = 5000 - for (let i = 0; i < textMessages.length; i += BATCH) { - const batch = textMessages.slice(i, i + BATCH) - db.transaction(() => { - for (const row of batch) { - insertFts.run(row.id, row.content) - } - }) - } -} - -function insertFtsEntries(db: DatabaseAdapter, onProgress?: (msg: string) => void): void { - try { - db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='message_fts'").get() - } catch { - return - } - - const hasFts = db - .prepare("SELECT COUNT(*) as cnt FROM sqlite_master WHERE type='table' AND name='message_fts'") - .get() as { cnt: number } - if (!hasFts || hasFts.cnt === 0) return - - onProgress?.('更新全文搜索索引...') - const maxRowid = (db.prepare('SELECT MAX(rowid) as m FROM message_fts').get() as { m: number | null })?.m ?? 0 - const newRows = db - .prepare("SELECT id, content FROM message WHERE type = 0 AND content IS NOT NULL AND content != '' AND id > ?") - .all(maxRowid) as Array<{ id: number; content: string }> - - if (newRows.length === 0) return - - const insertFts = db.prepare('INSERT INTO message_fts(rowid, content) VALUES (?, ?)') - db.transaction(() => { - for (const row of newRows) { - insertFts.run(row.id, row.content) - } - }) -} - -/** - * 执行导入（新建或增量） - */ -export async function importData( - dbManager: DatabaseManager, - data: ParsedData, - options?: ImportOptions & { onProgress?: (msg: string) => void } -): Promise { - const sessionId = options?.sessionId || generateSessionId() - const dbPath = dbManager.getDbPath(sessionId) - const exists = fs.existsSync(dbPath) - - if (options?.dryRun) { - if (exists) { - const db = dbManager.open(sessionId) - if (!db) - return { - success: false, - sessionId, - created: false, - messageCount: 0, - memberCount: 0, - duplicateCount: 0, - error: 'Failed to open database', - } - - const existingCount = (db.prepare('SELECT COUNT(*) as cnt FROM message').get() as { cnt: number }).cnt - return { - success: true, - sessionId, - created: false, - messageCount: data.messages.length, - memberCount: data.members.length, - duplicateCount: 0, - error: `Dry run: ${data.messages.length} messages to import into existing session (${existingCount} existing messages)`, - } - } - return { - success: true, - sessionId, - created: true, - messageCount: data.messages.length, - memberCount: data.members.length, - duplicateCount: 0, - error: `Dry run: would create new session with ${data.messages.length} messages`, - } - } - - try { - if (exists) { - const db = dbManager.openRawSessionDatabase(sessionId, { readonly: false }) - let result: { messageCount: number; memberCount: number; duplicateCount: number } - try { - result = incrementalImport(db, data, options?.onProgress) - } finally { - db.close() - } - dbManager.raiseCurrentChatDbCompatibilityGate() - return { success: true, sessionId, created: false, ...result } - } - - const db = dbManager.openRawSessionDatabase(sessionId, { create: true }) - let result: { messageCount: number; memberCount: number; duplicateCount: number } - try { - result = fullImport(db, data, options?.onProgress) - } finally { - db.close() - } - dbManager.raiseCurrentChatDbCompatibilityGate() - return { success: true, sessionId, created: true, ...result } - } catch (err: unknown) { - const message = err instanceof Error ? err.message : String(err) - if (!exists) { - dbManager.deleteSessionDatabaseFiles(sessionId) - } - return { - success: false, - sessionId, - created: false, - messageCount: 0, - memberCount: 0, - duplicateCount: 0, - error: message, - } - } -} diff --git a/apps/cli/src/import/index.ts b/apps/cli/src/import/index.ts index 8ddb62ef..c33e3f6f 100644 --- a/apps/cli/src/import/index.ts +++ b/apps/cli/src/import/index.ts @@ -1,9 +1,3 @@ -// Legacy ChatLab-only reader (kept for backward compatibility) -export { parseFile, detectFormat as detectChatLabFormat } from './chatlab-reader' -export type { ParsedData, ImportMeta, ImportMember, ImportMessage, ProgressCallback } from './chatlab-reader' -export { importData } from './importer' -export type { ImportResult, ImportOptions } from './importer' - // Full-format stream import via @openchatlab/parser + node-runtime streaming importer export { streamImport, diff --git a/apps/cli/src/import/stream-import.ts b/apps/cli/src/import/stream-import.ts index 909673f0..4c5138f6 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -53,6 +53,8 @@ export interface StreamImportOptions { chatIndex?: number nativeBinding?: string onProgress?: (progress: StreamImportProgress) => void + /** Fix the target session ID instead of auto-generating one. Used by sync/pull adapters. */ + sessionId?: string } function generateSessionId(): string { @@ -87,7 +89,7 @@ export async function streamImport( filePath: string, options?: StreamImportOptions ): Promise { - const { formatId, chatIndex, onProgress } = options || {} + const { formatId, chatIndex, onProgress, sessionId } = options || {} const formatOptions: Record = {} if (formatId) formatOptions.formatId = formatId @@ -132,7 +134,7 @@ export async function streamImport( : () => {} const deps = buildStreamImportDeps(dbManager, progressAdapter) - const result = await streamingImport(filePath, deps, formatOptions) + const result = await streamingImport(filePath, deps, formatOptions, sessionId) if (!result.success || !result.sessionId) return result try { diff --git a/apps/cli/src/sync/adapters.ts b/apps/cli/src/sync/adapters.ts index 05672ca3..fe3600bf 100644 --- a/apps/cli/src/sync/adapters.ts +++ b/apps/cli/src/sync/adapters.ts @@ -2,7 +2,7 @@ * Server-side implementations of @openchatlab/sync abstractions. * * NodeFetcher: uses Node.js fetch API - * DirectImporter: uses DatabaseManager + streamImport/importData + * DirectImporter: uses DatabaseManager + streamImport/incrementalImport * NoopNotifier: placeholder (future: SSE push) */ @@ -15,8 +15,7 @@ import { NOOP_LOGGER } from '@openchatlab/sync' import { buildPullUrl } from '@openchatlab/sync' import type { DatabaseManager } from '@openchatlab/node-runtime' import { DataDirCompatibilityError } from '@openchatlab/node-runtime' -import { parseFile } from '../import/chatlab-reader' -import { importData } from '../import/importer' +import { streamImport, incrementalImport } from '../import/stream-import' function getTempFilePath(ext: string): string { const id = crypto.randomBytes(8).toString('hex') @@ -106,29 +105,19 @@ export class DirectImporter implements DataImporter { private async incrementalImportFile(sessionId: string, tempFile: string): Promise { try { - const data = await parseFile(tempFile) - this.dbManager.close(sessionId) - const result = await importData(this.dbManager, data, { - sessionId, - }) + const result = await incrementalImport(this.dbManager, sessionId, tempFile) if (result.success) { + const newMessageCount = result.newMessageCount + const duplicateCount = result.batch?.duplicateCount ?? 0 this.logger.info( - `[DirectImporter] Incremental OK: +${result.messageCount} messages (${result.duplicateCount} duplicates skipped)` + `[DirectImporter] Incremental OK: +${newMessageCount} messages (${duplicateCount} duplicates skipped)` ) - return { - success: true, - newMessageCount: result.messageCount, - sessionId, - } + return { success: true, newMessageCount, sessionId } } - if ( - result.error?.includes('not found') || - result.error?.includes('session_not_found') || - result.error?.includes('no such table') - ) { + if (result.error === 'error.session_not_found' || result.error?.includes('no such table')) { return { success: false, newMessageCount: 0, sessionId, needFullResync: true } } @@ -141,17 +130,12 @@ export class DirectImporter implements DataImporter { private async fullImportFile(tempFile: string, externalId: string): Promise { try { - const data = await parseFile(tempFile) - - const result = await importData(this.dbManager, data, { - sessionId: externalId, - }) + const result = await streamImport(this.dbManager, tempFile, { sessionId: externalId }) if (result.success) { - this.logger.info( - `[DirectImporter] Full import OK: ${result.messageCount} messages, ${result.memberCount} members` - ) - return { success: true, newMessageCount: result.messageCount, sessionId: externalId } + const newMessageCount = result.diagnostics?.messagesWritten ?? 0 + this.logger.info(`[DirectImporter] Full import OK: +${newMessageCount} messages`) + return { success: true, newMessageCount, sessionId: result.sessionId ?? externalId } } return { success: false, newMessageCount: 0, error: result.error }