Function streamImport

apps/cli/src/import/stream-import.ts:87–152

(
  dbManager: DatabaseManager,
  filePath: string,
  options?: StreamImportOptions
)

 * with batched transactions, deferred indexes, and FTS.
 */
export async function streamImport(
  dbManager: DatabaseManager,
  filePath: string,
  options?: StreamImportOptions
): Promise<StreamImportResult> {
  const { formatId, chatIndex, onProgress, sessionId } = options || {}

  const formatOptions: Record<string, unknown> = {}
  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 'importing':
          case 'saving':
            stage = 'saving'
            pct = 80
            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 deps = buildStreamImportDeps(dbManager, progressAdapter)
  const result = await streamingImport(filePath, deps, formatOptions, sessionId)
  if (!result.success || !result.sessionId) return result

  try {
    dbManager.raiseCurrentChatDbCompatibilityGate()
  } catch (error) {
    deleteSessionDatabase(dbManager, result.sessionId)
    return {