diff --git a/apps/cli/src/import/stream-import.ts b/apps/cli/src/import/stream-import.ts index 4c5138f6..3521fae7 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -108,11 +108,14 @@ export async function streamImport( stage = 'parsing' pct = Math.min(Math.round(progress.percentage * 0.7), 70) break - case 'importing': case 'saving': stage = 'saving' pct = 80 break + case 'indexing': + stage = 'indexing' + pct = 90 + break case 'done': stage = 'done' pct = 100 diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index 802b6e00..4c7499a7 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -84,6 +84,13 @@ export interface StreamImportDeps { const BATCH_COMMIT_SIZE = 50000 const CHECKPOINT_INTERVAL = 200000 +/** + * Let the event loop process pending I/O before a long synchronous step. + * Without this, progress events written to an SSE socket stay buffered + * until the blocking work (index/FTS build) finishes. + */ +const yieldToEventLoop = () => new Promise((resolve) => setImmediate(resolve)) + function defaultGenerateSessionId(): string { const ts = Date.now() const rand = Math.random().toString(36).substring(2, 8) @@ -268,7 +275,7 @@ async function streamImportSingle( } onProgress({ - stage: 'importing', + stage: 'saving', bytesRead: 0, totalBytes: 0, messagesProcessed: totalMessageCount, @@ -466,13 +473,14 @@ async function streamImportSingle( // Flush nickname history in batch onProgress({ - stage: 'importing', + stage: 'saving', bytesRead: 0, totalBytes: 0, messagesProcessed: totalMessageCount, percentage: 100, message: '', }) + await yieldToEventLoop() logger?.perf('Writing nickname history', totalMessageCount) db.exec('BEGIN TRANSACTION') @@ -493,13 +501,14 @@ async function streamImportSingle( // Create indexes (deferred for performance) onProgress({ - stage: 'importing', + stage: 'indexing', bytesRead: 0, totalBytes: 0, messagesProcessed: totalMessageCount, percentage: 100, message: '', }) + await yieldToEventLoop() logger?.perf('Creating indexes', totalMessageCount) db.exec(CHAT_DB_INDEXES) logger?.perf('Indexes created', totalMessageCount) @@ -512,15 +521,16 @@ async function streamImportSingle( logger?.error('FTS index build failed (non-fatal)', ftsError instanceof Error ? ftsError : undefined) } - // Final WAL checkpoint + // Final WAL checkpoint + session index + post-import hook onProgress({ - stage: 'importing', + stage: 'indexing', bytesRead: 0, totalBytes: 0, messagesProcessed: totalMessageCount, percentage: 100, message: '', }) + await yieldToEventLoop() doCheckpoint() logger?.perf('WAL checkpoint done', totalMessageCount) diff --git a/packages/parser/src/native/chatlab-native.ts b/packages/parser/src/native/chatlab-native.ts index af785e11..cf22cffb 100644 --- a/packages/parser/src/native/chatlab-native.ts +++ b/packages/parser/src/native/chatlab-native.ts @@ -117,6 +117,10 @@ async function* parseChatLabNative( if (!batch) break messagesProcessed += batch.length yield { type: 'messages', data: batch.map(toParsedMessage) } + // The consumer typically does synchronous DB writes per batch; yield a + // macrotask so the event loop can flush pending I/O (e.g. SSE progress + // events) instead of being starved until all batches are delivered. + await new Promise((resolve) => setImmediate(resolve)) } const doneProgress = createProgress('done', totalBytes, totalBytes, messagesProcessed, '') diff --git a/packages/parser/src/native/weflow-native.ts b/packages/parser/src/native/weflow-native.ts index a383aa44..e5e7b26a 100644 --- a/packages/parser/src/native/weflow-native.ts +++ b/packages/parser/src/native/weflow-native.ts @@ -99,6 +99,10 @@ async function* parseWeFlowNative( if (!batch) break messagesProcessed += batch.length yield { type: 'messages', data: batch.map(toParsedMessage) } + // The consumer typically does synchronous DB writes per batch; yield a + // macrotask so the event loop can flush pending I/O (e.g. SSE progress + // events) instead of being starved until all batches are delivered. + await new Promise((resolve) => setImmediate(resolve)) } const doneProgress = createProgress('done', totalBytes, totalBytes, messagesProcessed, '') diff --git a/packages/parser/src/types.ts b/packages/parser/src/types.ts index 7eaa7b97..3ce80bb8 100644 --- a/packages/parser/src/types.ts +++ b/packages/parser/src/types.ts @@ -24,8 +24,9 @@ export interface ParsedMeta { */ export interface ParseProgress { /** 阶段 */ - // 导入流程会复用解析进度结构，因此这里补充导入阶段枚举。 - stage: 'detecting' | 'parsing' | 'importing' | 'saving' | 'done' | 'error' + // 导入流程会复用解析进度结构，因此这里补充导入阶段枚举： + // saving = 消息批量写库，indexing = 建索引/FTS/会话分段等后处理。 + stage: 'detecting' | 'parsing' | 'saving' | 'indexing' | 'done' | 'error' /** 已读取字节数 */ bytesRead: number /** 文件总字节数 */ diff --git a/src/i18n/locales/en-US/home.json b/src/i18n/locales/en-US/home.json index 95aab42d..bd97b4bd 100644 --- a/src/i18n/locales/en-US/home.json +++ b/src/i18n/locales/en-US/home.json @@ -23,6 +23,7 @@ "reading": "Reading file...", "parsing": "Parsing messages...", "saving": "Saving to database...", + "indexing": "Building search index...", "done": "Import complete", "error": "Import interrupted" }, diff --git a/src/i18n/locales/ja-JP/home.json b/src/i18n/locales/ja-JP/home.json index 4c14f94f..316b8f19 100644 --- a/src/i18n/locales/ja-JP/home.json +++ b/src/i18n/locales/ja-JP/home.json @@ -23,6 +23,7 @@ "reading": "ファイルを読み込み中...", "parsing": "メッセージを解析中...", "saving": "データベースに書き込み中...", + "indexing": "検索インデックスを構築中...", "done": "インポート完了", "error": "インポート中断" }, diff --git a/src/i18n/locales/zh-CN/home.json b/src/i18n/locales/zh-CN/home.json index 39de0792..0257e6cf 100644 --- a/src/i18n/locales/zh-CN/home.json +++ b/src/i18n/locales/zh-CN/home.json @@ -23,6 +23,7 @@ "reading": "正在读取文件...", "parsing": "正在解析消息...", "saving": "正在写入数据库...", + "indexing": "正在构建搜索索引...", "done": "导入完成", "error": "导入中断" }, diff --git a/src/i18n/locales/zh-TW/home.json b/src/i18n/locales/zh-TW/home.json index 44d6bbd6..5c4bf0c9 100644 --- a/src/i18n/locales/zh-TW/home.json +++ b/src/i18n/locales/zh-TW/home.json @@ -23,6 +23,7 @@ "reading": "正在讀取檔案...", "parsing": "正在解析訊息...", "saving": "正在寫入資料庫...", + "indexing": "正在建立搜尋索引...", "done": "匯入完成", "error": "匯入中斷" }, diff --git a/src/pages/home/components/import/ImportArea.vue b/src/pages/home/components/import/ImportArea.vue index 7d0971b5..477eab21 100644 --- a/src/pages/home/components/import/ImportArea.vue +++ b/src/pages/home/components/import/ImportArea.vue @@ -741,6 +741,8 @@ function getProgressText(): string { return t('home.import.progress.parsing') case 'saving': return t('home.import.progress.saving') + case 'indexing': + return t('home.import.progress.indexing') case 'done': return t('home.import.progress.done') case 'error':