feat(cli): enable automatic incremental imports

apps/cli/src/cli.ts

@@ -58,7 +58,7 @@ registerManifestCommand(program, getVersion())
 program
   .command('import <file>')
   .description('Import a chat history file (14+ formats: QQ/WeChat/Telegram/WhatsApp/LINE/Discord/Instagram, etc.)')
-  .option('--session-id <id>', 'Specify session ID (auto-generated if omitted)')
+  .option('--session-id <id>', 'Force an existing session target, or create with this ID if missing')
   .option('--format <id>', 'Specify format ID (skip auto-detection)')
   .action(async (file, options) => {
     if (!fs.existsSync(file)) {
@@ -66,7 +66,7 @@ program
       process.exit(1)
     }
 
-    const { streamImport, detectFormat } = await import('./import')
+    const { autoImport, detectFormat } = await import('./import')
     const { dbManager, pathProvider } = initRuntime()
     const nativeBinding = resolveNativeBinding()
     // 记录 Rust native parser 可用性（本次导入是否走 Rust 内核）
@@ -83,8 +83,9 @@ program
     if (format) console.log(`  Format: ${format.name} (${format.platform})`)
 
     try {
-      const result = await streamImport(dbManager, file, {
+      const result = await autoImport(dbManager, file, {
         formatId: options.format,
+        sessionId: options.sessionId,
         nativeBinding,
         onProgress: (p) => {
           process.stdout.write(`\r  ${p.stage}: ${p.progress}%`)
@@ -94,8 +95,10 @@ program
       if (result.success) {
         console.log(`\n\nImport succeeded!`)
         console.log(`  Session ID: ${result.sessionId}`)
-        console.log(`  Messages written: ${result.diagnostics?.messagesWritten ?? 0}`)
-        console.log(`  Messages skipped: ${result.diagnostics?.messagesSkipped ?? 0}`)
+        console.log(`  Mode: ${result.importMode === 'incremental' ? 'incremental' : 'created'}`)
+        if (result.matchedBy) console.log(`  Matched by: ${result.matchedBy}`)
+        console.log(`  New messages: ${result.newMessageCount ?? 0}`)
+        console.log(`  Duplicates skipped: ${result.duplicateCount ?? 0}`)
 
         if (result.sessionId) {
           try {
@@ -115,11 +118,11 @@ program
         }
       } else {
         console.error(`\n\nImport failed: ${result.error}`)
-        process.exit(1)
+        process.exitCode = 1
       }
     } catch (err) {
       console.error(`\n\nImport error: ${err instanceof Error ? err.message : err}`)
-      process.exit(1)
+      process.exitCode = 1
     } finally {
       dbManager.closeAll()
     }

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 { streamImport } from './stream-import'
+import { autoImport, streamImport } from './stream-import'
 
 const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
 
@@ -36,7 +36,7 @@ function writeTempChatFile(dir: string): string {
     filePath,
     JSON.stringify({
       chatlab: { version: '0.0.2', exportedAt: 1711468800 },
-      meta: { name: 'Test Chat', platform: 'qq', type: 'group' },
+      meta: { name: 'Test Chat', platform: 'qq', type: 'group', groupId: 'group-42' },
       members: [{ platformId: 'u1', accountName: 'Alice' }],
       // accountName is required by streaming-importer (skipped otherwise)
       messages: [{ sender: 'u1', accountName: 'Alice', timestamp: 1711468800, type: 0, content: 'hello' }],
@@ -91,3 +91,48 @@ test('streamImport re-checks data directory compatibility before raw database wr
   assert.match(result.error ?? '', /requires runtime version 0\.26\.0 or newer/)
   assert.equal(fs.readdirSync(path.join(root, 'data', 'databases')).filter((name) => name.endsWith('.db')).length, 0)
 })
+
+test('autoImport creates once and then incrementally imports the same stable chat', async () => {
+  const root = makeTempDir()
+  fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true })
+  const manager = new DatabaseManager(createPathProvider(root), {
+    nativeBinding,
+    runtime: { version: '0.25.1', kind: 'cli' },
+  })
+  const chatFile = writeTempChatFile(root)
+
+  const created = await autoImport(manager, chatFile, { nativeBinding })
+  const incremental = await autoImport(manager, chatFile, { nativeBinding })
+
+  assert.equal(created.success, true)
+  assert.equal(created.importMode, 'created')
+  assert.equal(incremental.success, true)
+  assert.equal(incremental.sessionId, created.sessionId)
+  assert.equal(incremental.importMode, 'incremental')
+  assert.equal(incremental.matchedBy, 'stable-id')
+  assert.equal(incremental.newMessageCount, 0)
+  assert.equal(incremental.duplicateCount, 1)
+  assert.equal(manager.listSessionIds().length, 1)
+})
+
+test('autoImport uses an explicit id for create and then forces incremental import', async () => {
+  const root = makeTempDir()
+  fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true })
+  const manager = new DatabaseManager(createPathProvider(root), {
+    nativeBinding,
+    runtime: { version: '0.25.1', kind: 'cli' },
+  })
+  const chatFile = writeTempChatFile(root)
+
+  const created = await autoImport(manager, chatFile, { sessionId: 'explicit', nativeBinding })
+  const incremental = await autoImport(manager, chatFile, { sessionId: 'explicit', nativeBinding })
+
+  assert.equal(created.success, true)
+  assert.equal(created.sessionId, 'explicit')
+  assert.equal(created.importMode, 'created')
+  assert.equal(incremental.success, true)
+  assert.equal(incremental.sessionId, 'explicit')
+  assert.equal(incremental.importMode, 'incremental')
+  assert.equal(incremental.newMessageCount, 0)
+  assert.equal(manager.listSessionIds().length, 1)
+})

apps/cli/src/import/index.ts

@@ -1,6 +1,7 @@
 // Full-format stream import via @openchatlab/parser + node-runtime streaming importer
 export {
   streamImport,
+  autoImport,
   incrementalImport,
   analyzeIncrementalImport,
   analyzeNewImport,
@@ -20,4 +21,5 @@ export type {
   IncrementalImportResult,
   IncrementalAnalyzeResult,
   AnalyzeNewImportResult,
+  AutoImportResult,
 } from './stream-import'

apps/cli/src/import/stream-import.test.ts

@@ -111,3 +111,27 @@ test('analyzeIncrementalImport propagates data directory compatibility errors',
     (error) => error instanceof DataDirCompatibilityError && error.code === 'DATA_DIR_REQUIRES_NEWER_RUNTIME'
   )
 })
+
+test('incrementalImport honors an explicitly selected parser format', async () => {
+  const root = makeTempDir()
+  const pathProvider = createPathProvider(root)
+  const manager = new DatabaseManager(pathProvider, {
+    nativeBinding,
+    runtime: { version: '0.25.1', kind: 'cli' },
+  })
+  const db = manager.openRawSessionDatabase('existing', { create: true, initializeChatTables: true })
+  db.prepare('INSERT INTO meta (name, platform, type, imported_at) VALUES (?, ?, ?, ?)').run(
+    'Existing Chat',
+    'qq',
+    'group',
+    1000
+  )
+  db.close()
+
+  const filePath = path.join(root, 'explicit-format.txt')
+  writeIncrementalJsonl(filePath)
+  const result = await incrementalImport(manager, 'existing', filePath, { formatId: 'chatlab-jsonl' })
+
+  assert.equal(result.success, true)
+  assert.equal(result.newMessageCount, 1)
+})

apps/cli/src/import/stream-import.ts

@@ -9,6 +9,7 @@
 import type { DatabaseManager } from '@openchatlab/node-runtime'
 import {
   DataDirCompatibilityError,
+  autoImportFile as sharedAutoImportFile,
   streamingImport,
   analyzeNewImport as sharedAnalyzeNewImport,
   analyzeIncrementalImport as sharedAnalyzeIncremental,
@@ -23,6 +24,7 @@ import type {
   IncrementalImportDeps,
   ImportOptions,
   AnalyzeNewImportResult,
+  AutoImportResult,
 } from '@openchatlab/node-runtime'
 import {
   detectFormat as parserDetectFormat,
@@ -76,6 +78,48 @@ function buildStreamImportDeps(dbManager: DatabaseManager, onProgress?: ImportPr
   }
 }
 
+function createProgressAdapter(onProgress?: (progress: StreamImportProgress) => void): ImportProgressCallback {
+  if (!onProgress) return () => {}
+  return (progress) => {
+    let stage: StreamImportProgress['stage'] = 'parsing'
+    let pct = 0
+    switch (progress.stage) {
+      case 'detecting':
+        stage = 'detecting'
+        pct = 5
+        break
+      case 'parsing':
+        stage = 'parsing'
+        pct = Math.min(Math.round(progress.percentage * 0.7), 70)
+        break
+      case 'saving':
+        stage = 'saving'
+        pct = 80
+        break
+      case 'indexing':
+        stage = 'indexing'
+        pct = 90
+        break
+      case 'done':
+        stage = 'done'
+        pct = 100
+        break
+      case 'error':
+        stage = 'error'
+        pct = 0
+        break
+    }
+    onProgress({
+      stage,
+      progress: pct,
+      message: progress.message || '',
+      bytesRead: progress.bytesRead,
+      totalBytes: progress.totalBytes,
+      messagesProcessed: progress.messagesProcessed,
+    })
+  }
+}
+
 function deleteSessionDatabase(dbManager: DatabaseManager, sessionId: string): void {
   dbManager.deleteSessionDatabaseFiles(sessionId)
 }
@@ -95,46 +139,7 @@ export async function streamImport(
   if (formatId) formatOptions.formatId = formatId
   if (chatIndex !== undefined) formatOptions.chatIndex = chatIndex
 
-  const progressAdapter: ImportProgressCallback = onProgress
-    ? (progress) => {
-        let stage: StreamImportProgress['stage'] = 'parsing'
-        let pct = 0
-        switch (progress.stage) {
-          case 'detecting':
-            stage = 'detecting'
-            pct = 5
-            break
-          case 'parsing':
-            stage = 'parsing'
-            pct = Math.min(Math.round(progress.percentage * 0.7), 70)
-            break
-          case 'saving':
-            stage = 'saving'
-            pct = 80
-            break
-          case 'indexing':
-            stage = 'indexing'
-            pct = 90
-            break
-          case 'done':
-            stage = 'done'
-            pct = 100
-            break
-          case 'error':
-            stage = 'error'
-            pct = 0
-            break
-        }
-        onProgress({
-          stage,
-          progress: pct,
-          message: progress.message || '',
-          bytesRead: progress.bytesRead,
-          totalBytes: progress.totalBytes,
-          messagesProcessed: progress.messagesProcessed,
-        })
-      }
-    : () => {}
+  const progressAdapter = createProgressAdapter(onProgress)
 
   const deps = buildStreamImportDeps(dbManager, progressAdapter)
   const result = await streamingImport(filePath, deps, formatOptions, sessionId)
@@ -154,6 +159,43 @@ export async function streamImport(
   return result
 }
 
+export async function autoImport(
+  dbManager: DatabaseManager,
+  filePath: string,
+  options?: StreamImportOptions
+): Promise<AutoImportResult> {
+  const { formatId, chatIndex, onProgress, sessionId } = options ?? {}
+  const formatOptions: Record<string, unknown> = {}
+  if (formatId) formatOptions.formatId = formatId
+  if (chatIndex !== undefined) formatOptions.chatIndex = chatIndex
+  const progressAdapter = createProgressAdapter(onProgress)
+
+  return sharedAutoImportFile(
+    filePath,
+    {
+      listSessionIds: () => dbManager.listSessionIds(),
+      openReadonly: (candidateSessionId) => dbManager.openRawSessionDatabase(candidateSessionId, { readonly: true }),
+      onProgress: progressAdapter,
+      sessionExists: (candidateSessionId) => dbManager.listSessionIds().includes(candidateSessionId),
+      createSession: (sourcePath, sourceFormatOptions, explicitSessionId) =>
+        streamImport(dbManager, sourcePath, {
+          ...options,
+          ...sourceFormatOptions,
+          sessionId: explicitSessionId,
+        }),
+      appendSession: (targetSessionId, sourcePath, sourceFormatOptions) =>
+        incrementalImport(dbManager, targetSessionId, sourcePath, {
+          ...sourceFormatOptions,
+          onProgress: progressAdapter,
+        }),
+    },
+    {
+      explicitSessionId: sessionId,
+      formatOptions,
+    }
+  )
+}
+
 // ==================== Incremental import ====================
 
 function buildIncrementalDeps(
@@ -249,4 +291,5 @@ export type {
   IncrementalAnalyzeResult,
   AnalyzeNewImportResult,
   ImportOptions,
+  AutoImportResult,
 }