refactor(import): migrate CLI sync/pull path to shared streaming importer

Removes the legacy apps/cli/src/import/importer.ts (hand-rolled FTS, dedup,
session-ID generation) and apps/cli/src/import/chatlab-reader.ts (buffered
in-memory JSON/JSONL parser). DirectImporter in sync/adapters.ts now calls
streamImport / incrementalImport from stream-import.ts, which use the
@openchatlab/parser streaming pipeline and node-runtime shared logic.

Changes:
- Add sessionId? to StreamImportOptions so callers can fix the target session
  ID instead of auto-generating one (needed by pull engine full-import path)
- Rewrite DirectImporter.incrementalImportFile and fullImportFile to use
  the shared importers; map IncrementalImportResult fields to ImportResult
- Remove legacy exports from apps/cli/src/import/index.ts
- Rewrite importer.test.ts to use streamImport + a temp ChatLab JSON file
  (same two data-dir compatibility gate assertions as before)
- Delete importer.ts and chatlab-reader.ts

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/cli/src/import/chatlab-reader.ts

@@ -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<ParsedData> {
-  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<string, unknown>) => ({
-    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<string, unknown>, 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<ParsedData> {
-  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<string, unknown>
-    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<string, unknown> | 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<string>()
-  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<ParsedData> {
-  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)
-  }
-}

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)

apps/cli/src/import/importer.ts

@@ -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<string>()
-
-  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<string>): 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<ImportResult> {
-  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,
-    }
-  }
-}

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,

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<StreamImportResult> {
-  const { formatId, chatIndex, onProgress } = options || {}
+  const { formatId, chatIndex, onProgress, sessionId } = options || {}
 
   const formatOptions: Record<string, unknown> = {}
   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 {

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<ImportResult> {
     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<ImportResult> {
     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 }