From 62b7104f1f3fe95620736a8540f461ef9f62c387 Mon Sep 17 00:00:00 2001 From: digua Date: Fri, 31 Jul 2026 17:28:26 +0800 Subject: [PATCH] feat(import): add safe batch coordinator --- apps/cli/src/http/routes/web/import.test.ts | 133 ++++++++ apps/cli/src/http/routes/web/import.ts | 91 +++++- apps/cli/src/import/index.ts | 3 + .../cli/src/import/stream-import-flow.test.ts | 123 ++++++- apps/cli/src/import/stream-import.ts | 250 ++++++++++---- apps/desktop/main/ipc/chat.ts | 36 ++ apps/desktop/main/worker/dbWorker.ts | 16 + apps/desktop/main/worker/import/index.ts | 3 + .../main/worker/import/streamImport.ts | 112 +++++-- apps/desktop/main/worker/workerManager.ts | 107 +++++- apps/desktop/preload/apis/chat.ts | 14 + apps/desktop/preload/index.d.ts | 24 ++ .../src/import/auto-import-batch.test.ts | 307 ++++++++++++++++++ .../src/import/auto-import-batch.ts | 187 +++++++++++ .../src/import/auto-import-matcher.test.ts | 43 ++- .../src/import/auto-import-matcher.ts | 68 +++- .../node-runtime/src/import/auto-importer.ts | 9 + .../src/import/batch-coordinator.test.ts | 132 ++++++++ .../src/import/batch-coordinator.ts | 128 ++++++++ packages/node-runtime/src/import/index.ts | 9 +- .../src/import/perf-logger.test.ts | 35 ++ .../node-runtime/src/import/perf-logger.ts | 222 +++++++------ packages/node-runtime/src/index.ts | 13 + scripts/bench-streaming-import.mts | 75 +++-- src/services/import/electron.ts | 76 +++++ src/services/import/fetch.ts | 131 ++++++++ .../import/importSourceAdapters.test.ts | 186 +++++++++++ src/services/import/types.ts | 27 ++ src/services/index.ts | 3 + src/stores/session-batch-import.test.ts | 54 ++- src/stores/session.ts | 61 ++++ 31 files changed, 2455 insertions(+), 223 deletions(-) create mode 100644 packages/node-runtime/src/import/auto-import-batch.test.ts create mode 100644 packages/node-runtime/src/import/auto-import-batch.ts create mode 100644 packages/node-runtime/src/import/batch-coordinator.test.ts create mode 100644 packages/node-runtime/src/import/batch-coordinator.ts create mode 100644 packages/node-runtime/src/import/perf-logger.test.ts diff --git a/apps/cli/src/http/routes/web/import.test.ts b/apps/cli/src/http/routes/web/import.test.ts index e417bf25..610f5dec 100644 --- a/apps/cli/src/http/routes/web/import.test.ts +++ b/apps/cli/src/http/routes/web/import.test.ts @@ -1,4 +1,5 @@ import assert from 'node:assert/strict' +import fs from 'node:fs' import { describe, it } from 'node:test' import Fastify from 'fastify' import multipart from '@fastify/multipart' @@ -60,7 +61,139 @@ function directoryMultipartPayload(): { payload: Buffer; contentType: string } { } } +function batchMultipartPayload(): { payload: Buffer; contentType: string } { + const boundary = '----chatlab-batch-import-route-test' + const field = [`--${boundary}`, 'Content-Disposition: form-data; name="sessionGapThreshold"', '', '7200'].join('\r\n') + const files = ['first.json', 'second.json'].map((filename, index) => + [ + `--${boundary}`, + `Content-Disposition: form-data; name="files"; filename="${filename}"`, + 'Content-Type: application/json', + '', + JSON.stringify({ index }), + ].join('\r\n') + ) + return { + payload: Buffer.from([field, ...files, `--${boundary}--`, ''].join('\r\n')), + contentType: `multipart/form-data; boundary=${boundary}`, + } +} + describe('CLI Web automatic import route', () => { + it('runs uploaded files through one backend batch coordinator', async () => { + const app = Fastify() + const calls: Array<{ ids: string[]; paths: string[]; threshold?: number; concurrency?: number }> = [] + try { + await app.register(multipart) + registerImportRoutes( + app, + {} as any, + { + runAutoImportBatch: async (items: any[], options: any) => { + calls.push({ + ids: items.map((item) => item.id), + paths: items.map((item) => item.filePath), + threshold: options.sessionGapThreshold, + concurrency: options.concurrency, + }) + options.onItemStart(items[0], 0) + options.onItemProgress(items[0], 0, { stage: 'parsing', percentage: 50, message: '' }) + const results = [ + { id: '0', status: 'success', result: { success: true, sessionId: 'session-1' } }, + { id: '1', status: 'failed', error: 'expected failure' }, + ] + options.onItemComplete(items[0], 0, results[0]) + options.onItemComplete(items[1], 1, results[1]) + return results + }, + } as any + ) + + const body = batchMultipartPayload() + const response = await app.inject({ + method: 'POST', + url: '/_web/import/batch', + headers: { 'content-type': body.contentType }, + payload: body.payload, + }) + + assert.equal(response.statusCode, 200) + assert.deepEqual( + calls.map(({ ids, threshold, concurrency }) => ({ ids, threshold, concurrency })), + [{ ids: ['0', '1'], threshold: 7200, concurrency: undefined }] + ) + assert.equal( + calls[0].paths.every((filePath) => !fs.existsSync(filePath)), + true + ) + assert.match(response.body, /event: batch-start/) + assert.match(response.body, /event: batch-progress/) + assert.match(response.body, /event: batch-complete/) + assert.match(response.body, /event: done/) + assert.match(response.body, /"sessionId":"session-1"/) + assert.match(response.body, /"error":"expected failure"/) + } finally { + await app.close() + } + }) + + it('cancels pending Web batch items without disconnecting the result stream', async () => { + const app = Fastify() + let notifyStarted: (() => void) | undefined + const started = new Promise((resolve) => { + notifyStarted = resolve + }) + try { + await app.register(multipart) + registerImportRoutes( + app, + {} as any, + { + runAutoImportBatch: async (items: any[], options: any) => { + notifyStarted?.() + await new Promise((resolve) => { + if (options.signal.aborted) resolve() + else options.signal.addEventListener('abort', () => resolve(), { once: true }) + }) + const results = [ + { id: '0', status: 'success', result: { success: true, sessionId: 'session-1' } }, + { id: '1', status: 'cancelled' }, + ] + options.onItemComplete(items[0], 0, results[0]) + options.onItemComplete(items[1], 1, results[1]) + return results + }, + } as any + ) + + const body = batchMultipartPayload() + const responsePromise = app.inject({ + method: 'POST', + url: '/_web/import/batch', + headers: { + 'content-type': body.contentType, + 'x-chatlab-import-batch-id': 'batch-cancel-test', + }, + payload: body.payload, + }) + await started + + const cancelResponse = await app.inject({ + method: 'POST', + url: '/_web/import/batch/batch-cancel-test/cancel', + }) + const response = await responsePromise + + assert.equal(cancelResponse.statusCode, 200) + assert.deepEqual(cancelResponse.json(), { success: true, active: true }) + assert.match(response.body, /"id":"0","status":"success"/) + assert.match(response.body, /"id":"1","status":"cancelled"/) + assert.match(response.body, /event: done/) + } finally { + await app.close() + } + }) + it('forwards parser options and preserves a zero-new incremental result in the done event', async () => { const app = Fastify() const calls: Array<{ formatId?: string; chatIndex?: number; sessionGapThreshold?: number }> = [] diff --git a/apps/cli/src/http/routes/web/import.ts b/apps/cli/src/http/routes/web/import.ts index b474b74f..4970e029 100644 --- a/apps/cli/src/http/routes/web/import.ts +++ b/apps/cli/src/http/routes/web/import.ts @@ -13,6 +13,7 @@ import { } from '@openchatlab/node-runtime' import { autoImport, + autoImportBatch, streamImport, incrementalImport, analyzeIncrementalImport, @@ -23,7 +24,12 @@ import { scanMultiChatFile, findEntryFileInDirectory, } from '../../../import' -import type { AutoImportResult, StreamImportOptions } from '../../../import' +import type { + AutoImportBatchOptions, + AutoImportBatchRequest, + AutoImportResult, + StreamImportOptions, +} from '../../../import' import { resolveNativeBinding } from './helpers' const ARCHIVE_UPLOAD_LIMIT = 50 * 1024 * 1024 * 1024 @@ -31,6 +37,10 @@ const ARCHIVE_UPLOAD_LIMIT = 50 * 1024 * 1024 * 1024 interface ImportRouteOptions { sourceManager?: ArchiveImportSourceManager runAutoImport?: (filePath: string, options: StreamImportOptions) => Promise + runAutoImportBatch?: ( + items: AutoImportBatchRequest[], + options: AutoImportBatchOptions + ) => ReturnType runPreparedImport?: ( manifestPath: string, onProgress: (progress: unknown) => void, @@ -65,8 +75,10 @@ export function registerImportRoutes( dbManager: DatabaseManager, options: ImportRouteOptions = {} ): void { + const activeBatchControllers = new Map() const sourceManager = options.sourceManager ?? new ArchiveImportSourceManager() const runAutoImport = options.runAutoImport ?? autoImport.bind(null, dbManager) + const runAutoImportBatch = options.runAutoImportBatch ?? autoImportBatch.bind(null, dbManager) const runPreparedImport = options.runPreparedImport ?? (async (manifestPath: string, onProgress: (progress: unknown) => void, sessionGapThreshold?: number) => { @@ -252,6 +264,83 @@ export function registerImportRoutes( } }) + server.post('/_web/import/batch', async (request, reply) => { + const parts = (request as any).parts() + if (!parts) return reply.code(400).send({ error: 'No files uploaded' }) + + const requestedBatchId = request.headers['x-chatlab-import-batch-id'] + const batchId = typeof requestedBatchId === 'string' && requestedBatchId ? requestedBatchId : randomUUID() + if (activeBatchControllers.has(batchId)) { + return reply.code(409).send({ error: 'Batch import id is already active' }) + } + + const tmpDir = createChatLabTempDir('imports', 'batch-') + const items: AutoImportBatchRequest[] = [] + let sessionGapThreshold: number | undefined + let completed = false + const controller = new AbortController() + activeBatchControllers.set(batchId, controller) + + try { + for await (const part of parts) { + if (part.type === 'field' && part.fieldname === 'sessionGapThreshold') { + sessionGapThreshold = parseOptionalInteger(part.value) + continue + } + if (part.type !== 'file') continue + const index = items.length + const safeName = path.basename(part.filename || `upload-${index}`) + const filePath = path.join(tmpDir, `${index}-${safeName}`) + await pipeline(part.file, fs.createWriteStream(filePath, { flags: 'wx' })) + items.push({ id: String(index), filePath }) + } + + if (items.length === 0) return reply.code(400).send({ error: 'No files uploaded' }) + + reply.raw.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }) + reply.raw.once('close', () => { + if (!completed) controller.abort() + }) + + const sendEvent = (event: string, eventData: unknown) => { + if (!reply.raw.destroyed) { + reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`) + } + } + const results = await runAutoImportBatch(items, { + sessionGapThreshold, + signal: controller.signal, + onItemStart: (_item, index) => sendEvent('batch-start', { index }), + onItemProgress: (_item, index, progress) => sendEvent('batch-progress', { index, progress }), + onItemComplete: (_item, index, result) => sendEvent('batch-complete', { index, result }), + }) + completed = true + sendEvent('done', results) + } catch (error) { + const payload = { error: error instanceof Error ? error.message : String(error) } + if (!reply.raw.headersSent) return reply.code(500).send(payload) + if (!reply.raw.destroyed) { + reply.raw.write(`event: error\ndata: ${JSON.stringify(payload)}\n\n`) + } + } finally { + completed = true + if (activeBatchControllers.get(batchId) === controller) activeBatchControllers.delete(batchId) + reply.raw.end() + cleanupTemp(tmpDir) + } + }) + + server.post<{ Params: { batchId: string } }>('/_web/import/batch/:batchId/cancel', async (request) => { + const controller = activeBatchControllers.get(request.params.batchId) + if (!controller) return { success: false, active: false } + controller.abort() + return { success: true, active: true } + }) + // ==================== Directory Import ==================== server.post('/_web/import-directory', async (request, reply) => { diff --git a/apps/cli/src/import/index.ts b/apps/cli/src/import/index.ts index 803c2f9f..5aeda5d1 100644 --- a/apps/cli/src/import/index.ts +++ b/apps/cli/src/import/index.ts @@ -2,6 +2,7 @@ export { streamImport, autoImport, + autoImportBatch, analyzeAutoImport, incrementalImport, analyzeIncrementalImport, @@ -24,4 +25,6 @@ export type { AnalyzeNewImportResult, AutoImportAnalysisResult, AutoImportResult, + AutoImportBatchRequest, + AutoImportBatchOptions, } from './stream-import' diff --git a/apps/cli/src/import/stream-import-flow.test.ts b/apps/cli/src/import/stream-import-flow.test.ts index c9872423..5043a07d 100644 --- a/apps/cli/src/import/stream-import-flow.test.ts +++ b/apps/cli/src/import/stream-import-flow.test.ts @@ -12,7 +12,7 @@ import { readDataDirCompatibilityMeta, withDataDirImportLock, } from '@openchatlab/node-runtime' -import { analyzeAutoImport, autoImport, streamImport } from './stream-import' +import { analyzeAutoImport, autoImport, autoImportBatch, streamImport } from './stream-import' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -67,6 +67,20 @@ function writeTempDuplicateChatFile(dir: string): string { return filePath } +function writeBatchChatFile(dir: string, filename: string, groupId: string, content: string): string { + const filePath = path.join(dir, filename) + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 1711468800 }, + meta: { name: groupId, platform: 'qq', type: 'group', groupId }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + messages: [{ sender: 'u1', accountName: 'Alice', timestamp: 1711468800, type: 0, content }], + }) + ) + 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 }) @@ -151,6 +165,113 @@ test('autoImport creates once and then incrementally imports the same stable cha assert.equal(manager.listSessionIds().length, 1) }) +test('autoImportBatch holds one lock, coalesces the same target, and preserves independent results', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) + const manager = new DatabaseManager(createPathProvider(root), { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + const first = writeBatchChatFile(root, 'group-a-1.json', 'group-a', 'first') + const second = writeBatchChatFile(root, 'group-a-2.json', 'group-a', 'second') + const independent = writeBatchChatFile(root, 'group-b.json', 'group-b', 'independent') + + const results = await autoImportBatch( + manager, + [ + { id: 'a-1', filePath: first }, + { id: 'b', filePath: independent }, + { id: 'a-2', filePath: second }, + ], + { concurrency: 2, sessionGapThreshold: 7200 } + ) + + assert.deepEqual( + results.map((result) => result.status), + ['success', 'success', 'success'] + ) + assert.equal(results[0].status === 'success' && results[0].result.importMode, 'created') + assert.equal(results[2].status === 'success' && results[2].result.importMode, 'incremental') + assert.equal( + results[0].status === 'success' && + results[2].status === 'success' && + results[0].result.sessionId === results[2].result.sessionId, + true + ) + assert.equal(manager.listSessionIds().length, 2) + + const sameTargetId = results[0].status === 'success' ? results[0].result.sessionId! : '' + const db = manager.openRawSessionDatabase(sameTargetId, { readonly: true }) + const row = db + .prepare( + `SELECT + (SELECT COUNT(*) FROM message) AS messages, + (SELECT session_gap_threshold FROM meta LIMIT 1) AS gapThreshold` + ) + .get() as { messages: number; gapThreshold: number } + db.close() + assert.deepEqual(row, { messages: 2, gapThreshold: 7200 }) +}) + +test('autoImportBatch rejects an external writer and cancels work that has not started', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) + const manager = new DatabaseManager(createPathProvider(root), { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + const file = writeBatchChatFile(root, 'locked.json', 'locked', 'locked') + const lockedResults = await withDataDirImportLock(manager.getUserDataDir(), () => + autoImportBatch(manager, [{ id: 'locked', filePath: file }]) + ) + assert.deepEqual(lockedResults, [{ id: 'locked', status: 'failed', error: IMPORT_IN_PROGRESS_ERROR_KEY }]) + + const controller = new AbortController() + const items = Array.from({ length: 5 }, (_, index) => ({ + id: String(index), + filePath: writeBatchChatFile(root, `cancel-${index}.json`, `cancel-${index}`, `message-${index}`), + sessionId: `cancel-${index}`, + })) + let completed = 0 + const cancelledResults = await autoImportBatch(manager, items, { + concurrency: 1, + signal: controller.signal, + onItemComplete() { + completed++ + if (completed === 1) controller.abort() + }, + }) + + assert.equal(cancelledResults[0].status, 'success') + assert.deepEqual( + cancelledResults.slice(1).map((result) => result.status), + ['cancelled', 'cancelled', 'cancelled', 'cancelled'] + ) + assert.equal(manager.listSessionIds().length, 1) +}) + +test('autoImportBatch removes newly created sessions when the shared compatibility gate fails', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) + const manager = new DatabaseManager(createPathProvider(root), { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + manager.raiseCurrentChatDbCompatibilityGate = () => { + throw new Error('compatibility gate unavailable') + } + const file = writeBatchChatFile(root, 'gate.json', 'gate', 'message') + + const results = await autoImportBatch(manager, [{ id: 'gate', filePath: file, sessionId: 'gate-session' }]) + + assert.equal(results[0].status, 'failed') + assert.equal(results[0].status === 'failed' && results[0].error, 'compatibility gate unavailable') + assert.equal(fs.existsSync(path.join(root, 'data', 'databases', 'gate-session.db')), false) +}) + test('analyzeAutoImport previews a new session without writing a database', async (t) => { const root = makeTempDir() t.after(() => fs.rmSync(root, { recursive: true, force: true })) diff --git a/apps/cli/src/import/stream-import.ts b/apps/cli/src/import/stream-import.ts index 3f9f882d..513201e0 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -11,7 +11,9 @@ import { DataDirCompatibilityError, IMPORT_IN_PROGRESS_ERROR_KEY, ImportInProgressError, + resolveDefaultBatchConcurrency, analyzeAutoImportFile as sharedAnalyzeAutoImportFile, + autoImportBatch as sharedAutoImportBatch, autoImportFile as sharedAutoImportFile, streamingImport, analyzeNewImport as sharedAnalyzeNewImport, @@ -29,6 +31,7 @@ import type { ImportOptions, AnalyzeNewImportResult, AutoImportAnalysisResult, + AutoImportBatchItemResult, AutoImportResult, } from '@openchatlab/node-runtime' import { @@ -91,43 +94,45 @@ function buildStreamImportDeps( 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, - }) + return (progress) => onProgress(createProgressAdapterValue(progress)) +} + +function createProgressAdapterValue(progress: Parameters[0]): StreamImportProgress { + 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 + } + return { + stage, + progress: pct, + message: progress.message || '', + bytesRead: progress.bytesRead, + totalBytes: progress.totalBytes, + messagesProcessed: progress.messagesProcessed, } } @@ -142,7 +147,8 @@ function deleteSessionDatabase(dbManager: DatabaseManager, sessionId: string): v async function streamImportUnlocked( dbManager: DatabaseManager, filePath: string, - options?: StreamImportOptions + options?: StreamImportOptions, + updateCompatibilityGate = true ): Promise { const { formatId, chatIndex, sessionGapThreshold, onProgress, sessionId } = options || {} @@ -156,6 +162,8 @@ async function streamImportUnlocked( const result = await streamingImport(filePath, deps, formatOptions, sessionId) if (!result.success || !result.sessionId) return result + if (!updateCompatibilityGate) return result + try { dbManager.raiseCurrentChatDbCompatibilityGate() } catch (error) { @@ -191,6 +199,24 @@ export async function autoImport( dbManager: DatabaseManager, filePath: string, options?: StreamImportOptions +): Promise { + try { + return await withDataDirImportLock(dbManager.getUserDataDir(), () => + autoImportUnlocked(dbManager, filePath, options) + ) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY } + } + throw error + } +} + +async function autoImportUnlocked( + dbManager: DatabaseManager, + filePath: string, + options?: StreamImportOptions, + updateCompatibilityGate = true ): Promise { const { formatId, chatIndex, onProgress, sessionId } = options ?? {} const formatOptions: Record = {} @@ -198,37 +224,147 @@ export async function autoImport( 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) => + streamImportUnlocked( + dbManager, + sourcePath, + { + ...options, + ...sourceFormatOptions, + sessionId: explicitSessionId, + }, + updateCompatibilityGate + ), + appendSession: (targetSessionId, sourcePath, sourceFormatOptions) => + incrementalImportUnlocked( + dbManager, + targetSessionId, + sourcePath, + { + ...sourceFormatOptions, + onProgress: progressAdapter, + }, + updateCompatibilityGate + ), + }, + { + explicitSessionId: sessionId, + formatOptions, + } + ) +} + +export interface AutoImportBatchRequest { + id: string + filePath: string + formatId?: string + chatIndex?: number + sessionId?: string +} + +export interface AutoImportBatchOptions { + concurrency?: number + sessionGapThreshold?: number + signal?: AbortSignal + onItemStart?: (item: AutoImportBatchRequest, index: number) => void + onItemProgress?: (item: AutoImportBatchRequest, index: number, progress: StreamImportProgress) => void + onItemComplete?: (item: AutoImportBatchRequest, index: number, result: AutoImportBatchItemResult) => void +} + +export async function autoImportBatch( + dbManager: DatabaseManager, + items: AutoImportBatchRequest[], + options: AutoImportBatchOptions = {} +): Promise { try { - return await withDataDirImportLock(dbManager.getUserDataDir(), () => - sharedAutoImportFile( - filePath, + return await withDataDirImportLock(dbManager.getUserDataDir(), async () => { + const results = await sharedAutoImportBatch( + items.map((item) => ({ + id: item.id, + filePath: item.filePath, + options: { + explicitSessionId: item.sessionId, + formatOptions: { + ...(item.formatId ? { formatId: item.formatId } : {}), + ...(item.chatIndex !== undefined ? { chatIndex: item.chatIndex } : {}), + }, + }, + })), { listSessionIds: () => dbManager.listSessionIds(), openReadonly: (candidateSessionId) => dbManager.openRawSessionDatabase(candidateSessionId, { readonly: true }), - onProgress: progressAdapter, sessionExists: (candidateSessionId) => dbManager.listSessionIds().includes(candidateSessionId), createSession: (sourcePath, sourceFormatOptions, explicitSessionId) => - streamImportUnlocked(dbManager, sourcePath, { - ...options, - ...sourceFormatOptions, - sessionId: explicitSessionId, - }), + streamImportUnlocked( + dbManager, + sourcePath, + { + ...sourceFormatOptions, + sessionId: explicitSessionId, + sessionGapThreshold: options.sessionGapThreshold, + }, + false + ), appendSession: (targetSessionId, sourcePath, sourceFormatOptions) => - incrementalImportUnlocked(dbManager, targetSessionId, sourcePath, { - ...sourceFormatOptions, - onProgress: progressAdapter, - }), + incrementalImportUnlocked( + dbManager, + targetSessionId, + sourcePath, + { + ...sourceFormatOptions, + }, + false + ), }, { - explicitSessionId: sessionId, - formatOptions, + concurrency: options.concurrency ?? resolveDefaultBatchConcurrency(items.length), + signal: options.signal, + onItemStart: (_item, index) => options.onItemStart?.(items[index], index), + onItemProgress: (_item, index, progress) => + options.onItemProgress?.(items[index], index, createProgressAdapterValue(progress)), + onItemComplete: (_item, index, result) => options.onItemComplete?.(items[index], index, result), } ) - ) + + if (results.some((result) => result.status === 'success')) { + try { + dbManager.raiseCurrentChatDbCompatibilityGate() + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + for (const result of results) { + if (result.status === 'success' && result.result.importMode === 'created' && result.result.sessionId) { + deleteSessionDatabase(dbManager, result.result.sessionId) + } + } + return results.map((result) => + result.status === 'success' + ? { + id: result.id, + status: 'failed' as const, + error: message, + result: { ...result.result, success: false, error: message }, + } + : result + ) + } + } + return results + }) } catch (error) { if (error instanceof ImportInProgressError) { - return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY } + return items.map((item) => ({ + id: item.id, + status: 'failed' as const, + error: IMPORT_IN_PROGRESS_ERROR_KEY, + })) } throw error } @@ -296,7 +432,8 @@ async function incrementalImportUnlocked( dbManager: DatabaseManager, sessionId: string, filePath: string, - options?: ImportOptions & { onProgress?: ImportProgressCallback } + options?: ImportOptions & { onProgress?: ImportProgressCallback }, + updateCompatibilityGate = true ): Promise { const { onProgress, ...importOpts } = options || {} let compatibilityError: DataDirCompatibilityError | null = null @@ -310,6 +447,7 @@ async function incrementalImportUnlocked( ) if (compatibilityError) throw compatibilityError if (!result.success) return result + if (!updateCompatibilityGate) return result try { dbManager.raiseCurrentChatDbCompatibilityGate() diff --git a/apps/desktop/main/ipc/chat.ts b/apps/desktop/main/ipc/chat.ts index 999ffca2..53aa9fd1 100644 --- a/apps/desktop/main/ipc/chat.ts +++ b/apps/desktop/main/ipc/chat.ts @@ -233,6 +233,42 @@ export function registerChatHandlers(ctx: IpcContext): void { } }) + ipcMain.handle( + 'chat:importBatch', + async ( + _, + batchId: string, + items: Array<{ id: string; filePath: string }>, + options?: { concurrency?: number; sessionGapThreshold?: number } + ) => { + try { + const results = await worker.autoImportBatch( + batchId, + items, + (progress) => { + win.webContents.send('chat:importBatchProgress', { batchId, ...progress }) + }, + options + ) + if (results.some((result) => result.status === 'success')) { + win.webContents.send('api:importCompleted') + } + return results + } catch (error) { + return items.map((item) => ({ + id: item.id, + status: 'failed' as const, + error: error instanceof Error ? error.message : String(error), + })) + } + } + ) + + ipcMain.handle('chat:cancelImportBatch', async (_, batchId: string) => { + await worker.cancelAutoImportBatch(batchId) + return { success: true } + }) + // ==================== 增量导入 ==================== ipcMain.handle('chat:analyzeIncrementalImport', async (_, sessionId: string, filePath: string) => { diff --git a/apps/desktop/main/worker/dbWorker.ts b/apps/desktop/main/worker/dbWorker.ts index 2cce168e..9eca39b5 100644 --- a/apps/desktop/main/worker/dbWorker.ts +++ b/apps/desktop/main/worker/dbWorker.ts @@ -81,6 +81,7 @@ import { import { streamImport, autoImport, + autoImportBatch, streamParseFileInfo, analyzeIncrementalImport, incrementalImport, @@ -89,6 +90,8 @@ import { pushImport, } from './import' +const importBatchControllers = new Map() + initDbDir(workerData.dbDir, workerData.cacheDir, workerData.tempDir, workerData.nativeBinding, workerData.logsDir) // 初始化 NLP 词库目录 @@ -191,6 +194,10 @@ const syncHandlers: Record any> = { closeAllDatabases() return true }, + cancelAutoImportBatch: (p) => { + importBatchControllers.get(p.batchId)?.abort() + return true + }, // 成员管理 getMembers: (p) => getMembers(p.sessionId), @@ -253,6 +260,15 @@ const asyncHandlers: Record Promise // 流式导入 streamImport: (p, id) => streamImport(p.filePath, id, p.formatOptions, p.externalSessionId, p.sessionGapThreshold), autoImport: (p, id) => autoImport(p.filePath, id, p.formatOptions, p.explicitSessionId, p.sessionGapThreshold), + autoImportBatch: async (p, id) => { + const controller = new AbortController() + importBatchControllers.set(p.batchId, controller) + try { + return await autoImportBatch(p.items, id, p.concurrency, p.sessionGapThreshold, controller.signal) + } finally { + importBatchControllers.delete(p.batchId) + } + }, analyzePushImport: (p) => analyzePushImport(p.sessionId, p.payload), pushImport: (p) => pushImport(p.sessionId, p.payload), // 流式解析文件信息（用于合并预览） diff --git a/apps/desktop/main/worker/import/index.ts b/apps/desktop/main/worker/import/index.ts index d1b4835d..39d5d966 100644 --- a/apps/desktop/main/worker/import/index.ts +++ b/apps/desktop/main/worker/import/index.ts @@ -7,10 +7,13 @@ export { streamImport, autoImport, + autoImportBatch, streamParseFileInfo, analyzeNewImport, type StreamImportResult, type AutoImportResult, + type AutoImportBatchItemResult, + type WorkerAutoImportBatchItem, type StreamParseFileInfoResult, type AnalyzeNewImportResult, } from './streamImport' diff --git a/apps/desktop/main/worker/import/streamImport.ts b/apps/desktop/main/worker/import/streamImport.ts index 5f2fb0cc..a5ef49d3 100644 --- a/apps/desktop/main/worker/import/streamImport.ts +++ b/apps/desktop/main/worker/import/streamImport.ts @@ -11,28 +11,31 @@ import * as path from 'path' import type Database from 'better-sqlite3' import { BetterSqliteAdapter, + autoImportBatch as sharedAutoImportBatch, autoImportFile as sharedAutoImportFile, streamingImport, analyzeNewImport as sharedAnalyzeNewImport, streamParseFileInfo as sharedStreamParseFileInfo, TEMP_DB_SCHEMA, computeAndSetOverviewCache, + createImportPerfLogger, deleteSessionCache, - getCurrentLogFile, - logError, - logInfo, - logPerf, - logPerfDetail, - logSummary, - resetPerfLog, } from '@openchatlab/node-runtime' -import type { AutoImportResult, StreamImportDeps, StreamImportResult, ImportLogger } from '@openchatlab/node-runtime' +import type { + AutoImportBatchItemResult, + AutoImportResult, + StreamImportDeps, + StreamImportResult, + ImportLogger, +} from '@openchatlab/node-runtime' import { sendProgress, generateSessionId, getDbPath, createDatabaseWithoutIndexes } from './utils' import { incrementalImport } from './incrementalImport' -import { getCacheDir, getDbDir, getTempDir, openRawDatabase, initPerfLog } from '../core' +import { getCacheDir, getDbDir, getLogsDir, getTempDir, openRawDatabase } from '../core' +import { getImportLogDir } from '../core/perfLogPath' export type { StreamImportResult } export type { AutoImportResult } +export type { AutoImportBatchItemResult } export type { AnalyzeNewImportResult, StreamParseFileInfoResult } from '@openchatlab/node-runtime' export type { SkipReasons, ImportDiagnostics } from '@openchatlab/node-runtime' @@ -54,16 +57,7 @@ function createTempDatabase(dbPath: string): Database.Database { } function buildElectronLogger(): ImportLogger { - return { - info: logInfo, - error: (message: string, err?: Error) => logError(message, err), - perf: (label: string, messageCount: number, batchSize?: number) => logPerf(label, messageCount, batchSize), - perfDetail: logPerfDetail, - summary: logSummary, - reset: resetPerfLog, - init: initPerfLog, - getCurrentLogFile, - } + return createImportPerfLogger(getImportLogDir(getLogsDir())) } function buildStreamImportDeps(requestId: string, sessionGapThreshold?: number): StreamImportDeps { @@ -155,6 +149,86 @@ export async function autoImport( ) } +export interface WorkerAutoImportBatchItem { + id: string + filePath: string + formatOptions?: Record + explicitSessionId?: string +} + +export async function autoImportBatch( + items: WorkerAutoImportBatchItem[], + requestId: string, + concurrency: number, + sessionGapThreshold?: number, + signal?: AbortSignal +): Promise { + return sharedAutoImportBatch( + items.map((item) => ({ + id: item.id, + filePath: item.filePath, + options: { + explicitSessionId: item.explicitSessionId, + formatOptions: item.formatOptions, + }, + })), + { + listSessionIds: () => + fs.existsSync(getDbDir()) + ? fs + .readdirSync(getDbDir()) + .filter((name) => name.endsWith('.db')) + .map((name) => name.slice(0, -3)) + : [], + openReadonly: (sessionId) => new BetterSqliteAdapter(openRawDatabase(getDbPath(sessionId), { readonly: true })), + sessionExists: (sessionId) => fs.existsSync(getDbPath(sessionId)), + createSession: (sourcePath, sourceFormatOptions, sessionId) => + streamImport(sourcePath, requestId, sourceFormatOptions, sessionId, sessionGapThreshold), + appendSession: (sessionId, sourcePath, sourceFormatOptions) => + incrementalImport(sessionId, sourcePath, requestId, { + formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, + chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, + }), + }, + { + concurrency, + signal, + onItemStart: (_item, index) => { + sendProgress(requestId, { + stage: 'detecting', + percentage: 0, + message: '', + bytesRead: 0, + totalBytes: 0, + messagesProcessed: 0, + batchIndex: index, + batchEvent: 'start', + } as Parameters[1]) + }, + onItemProgress: (_item, index, progress) => { + sendProgress(requestId, { + ...progress, + batchIndex: index, + batchEvent: 'progress', + } as Parameters[1]) + }, + onItemComplete: (_item, index, result) => { + sendProgress(requestId, { + stage: 'done', + percentage: 100, + message: '', + bytesRead: 0, + totalBytes: 0, + messagesProcessed: 0, + batchIndex: index, + batchEvent: 'complete', + batchResult: result, + } as Parameters[1]) + }, + } + ) +} + /** * Dry-run analysis: parse without writing to DB. */ diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index 04ab6b6d..4c4ee74a 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/apps/desktop/main/worker/workerManager.ts @@ -18,9 +18,15 @@ import { IMPORT_IN_PROGRESS_ERROR_KEY, ImportInProgressError, raiseChatDbCompatibilityGate, + resolveDefaultBatchConcurrency, withDataDirImportLock, } from '@openchatlab/node-runtime' -import type { PushImportAnalysisOutcome, PushImportOutcome, PushImportPayload } from '@openchatlab/node-runtime' +import type { + AutoImportBatchItemResult, + PushImportAnalysisOutcome, + PushImportOutcome, + PushImportPayload, +} from '@openchatlab/node-runtime' import { getWorkerRequestTimeoutMs, isRestartableReadOnlyRequestType } from './workerTimeoutPolicy' interface WorkerRequestOptions { @@ -37,7 +43,7 @@ const pendingRequests = new Map< { resolve: (value: any) => void reject: (error: Error) => void - timeout: ReturnType + timeout?: ReturnType restartOnTimeout: boolean onProgress?: (progress: ParseProgress) => void // 进度回调 } @@ -55,7 +61,7 @@ function hasProgressRequest(): boolean { function rejectAllPending(error: Error): void { for (const [id, pending] of pendingRequests.entries()) { - clearTimeout(pending.timeout) + if (pending.timeout) clearTimeout(pending.timeout) pending.reject(error) pendingRequests.delete(id) } @@ -163,7 +169,7 @@ export function initWorker(): void { // 处理完成或错误消息 pendingRequests.delete(id) - clearTimeout(pending.timeout) + if (pending.timeout) clearTimeout(pending.timeout) if (success) { pending.resolve(result) @@ -243,7 +249,7 @@ function sendToWorkerWithProgress( type: string, payload: any, onProgress?: (progress: ParseProgress) => void, - timeoutMs: number = 600000 // 默认 10 分钟超时 + timeoutMs: number | null = 600000 // 默认 10 分钟超时；null 表示由调用方持有直至 Worker 完成 ): Promise { return new Promise((resolve, reject) => { if (!worker) { @@ -258,12 +264,15 @@ function sendToWorkerWithProgress( const requestWorker = worker! const id = `req_${++requestIdCounter}` - const timeout = setTimeout(() => { - if (pendingRequests.has(id)) { - pendingRequests.delete(id) - reject(new Error(`Worker request timeout: ${type}`)) - } - }, timeoutMs) + const timeout = + timeoutMs === null + ? undefined + : setTimeout(() => { + if (pendingRequests.has(id)) { + pendingRequests.delete(id) + reject(new Error(`Worker request timeout: ${type}`)) + } + }, timeoutMs) pendingRequests.set(id, { resolve, reject, timeout, restartOnTimeout: false, onProgress }) @@ -619,6 +628,82 @@ export async function autoImport( } } +export interface DesktopAutoImportBatchItem { + id: string + filePath: string + formatOptions?: Record + explicitSessionId?: string +} + +export interface DesktopAutoImportBatchProgress extends ParseProgress { + batchIndex: number + batchEvent: 'start' | 'progress' | 'complete' + batchResult?: AutoImportBatchItemResult +} + +export async function autoImportBatch( + batchId: string, + items: DesktopAutoImportBatchItem[], + onProgress?: (progress: DesktopAutoImportBatchProgress) => void, + options?: { concurrency?: number; sessionGapThreshold?: number } +): Promise { + try { + return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => { + assertDataDirCompatibleNow() + const results = await sendToWorkerWithProgress( + 'autoImportBatch', + { + batchId, + items, + concurrency: options?.concurrency ?? resolveDefaultBatchConcurrency(items.length), + sessionGapThreshold: options?.sessionGapThreshold, + }, + onProgress as (progress: ParseProgress) => void, + null + ) + if (!results.some((result) => result.status === 'success')) return results + + try { + raiseChatDbCompatibilityGate(getPathProvider(), { + version: getDesktopAppVersion(app.getVersion()), + kind: 'desktop', + }) + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + for (const result of results) { + if (result.status === 'success' && result.result.importMode === 'created' && result.result.sessionId) { + deleteImportedSession(result.result.sessionId) + } + } + return results.map((result) => + result.status === 'success' + ? { + id: result.id, + status: 'failed' as const, + error: message, + result: { ...result.result, success: false, error: message }, + } + : result + ) + } + return results + }) + } catch (error) { + if (error instanceof ImportInProgressError) { + return items.map((item) => ({ + id: item.id, + status: 'failed', + error: IMPORT_IN_PROGRESS_ERROR_KEY, + })) + } + throw error + } +} + +export async function cancelAutoImportBatch(batchId: string): Promise { + await sendToWorker('cancelAutoImportBatch', { batchId }) +} + /** * 在 worker 中执行共享 Push Import 写库语义；主进程持有全局导入锁， * 并在 worker 成功后完成数据目录兼容门禁更新。 diff --git a/apps/desktop/preload/apis/chat.ts b/apps/desktop/preload/apis/chat.ts index c5f10663..6325ae65 100644 --- a/apps/desktop/preload/apis/chat.ts +++ b/apps/desktop/preload/apis/chat.ts @@ -59,6 +59,20 @@ export const chatApi = { importWithOptions: (filePath: string, formatOptions: Record): Promise => ipcRenderer.invoke('chat:importWithOptions', filePath, formatOptions), + importBatch: ( + batchId: string, + items: Array<{ id: string; filePath: string }>, + options?: { concurrency?: number; sessionGapThreshold?: number } + ) => ipcRenderer.invoke('chat:importBatch', batchId, items, options), + + cancelImportBatch: (batchId: string) => ipcRenderer.invoke('chat:cancelImportBatch', batchId), + + onImportBatchProgress: (callback: (progress: Record) => void) => { + const handler = (_event: Electron.IpcRendererEvent, progress: Record) => callback(progress) + ipcRenderer.on('chat:importBatchProgress', handler) + return () => ipcRenderer.removeListener('chat:importBatchProgress', handler) + }, + scanMultiChatFile: ( filePath: string ): Promise<{ diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index 9374f3da..bf9b1ef9 100644 --- a/apps/desktop/preload/index.d.ts +++ b/apps/desktop/preload/index.d.ts @@ -88,6 +88,30 @@ interface ChatApi { import: (filePath: string) => Promise importDirectory: (dirPath: string, options?: Record) => Promise importWithOptions: (filePath: string, formatOptions: Record) => Promise + importBatch: ( + batchId: string, + items: Array<{ id: string; filePath: string }>, + options?: { concurrency?: number; sessionGapThreshold?: number } + ) => Promise< + Array<{ + id: string + status: 'success' | 'failed' | 'cancelled' + result?: ChatImportResult + error?: string + }> + > + cancelImportBatch: (batchId: string) => Promise<{ success: boolean }> + onImportBatchProgress: ( + callback: (progress: { + batchId: string + batchIndex: number + batchEvent: 'start' | 'progress' | 'complete' + stage: string + percentage: number + message?: string + batchResult?: unknown + }) => void + ) => () => void scanMultiChatFile: (filePath: string) => Promise<{ success: boolean chats: Array<{ index: number; name: string; type: string; id: number; messageCount: number }> diff --git a/packages/node-runtime/src/import/auto-import-batch.test.ts b/packages/node-runtime/src/import/auto-import-batch.test.ts new file mode 100644 index 00000000..b2a3e123 --- /dev/null +++ b/packages/node-runtime/src/import/auto-import-batch.test.ts @@ -0,0 +1,307 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { setTimeout as delay } from 'node:timers/promises' +import type { AutoImportDeps } from './auto-importer' +import { autoImportBatch } from './auto-import-batch' +import type { AutoImportTargetPlan } from './auto-import-matcher' + +test('parallelizes distinct explicit targets and serializes identical targets', async () => { + const activeTargets = new Set() + const existingSessions = new Set(['existing']) + let active = 0 + let maxActive = 0 + const creates: string[] = [] + const appends: Array<{ sessionId: string; filePath: string }> = [] + + const enter = async (target: string) => { + assert.equal(activeTargets.has(target), false) + activeTargets.add(target) + active++ + maxActive = Math.max(maxActive, active) + await delay(4) + active-- + activeTargets.delete(target) + } + const deps: AutoImportDeps = { + listSessionIds: () => [...existingSessions], + openReadonly: () => { + throw new Error('preflight is stubbed') + }, + sessionExists: (sessionId) => existingSessions.has(sessionId), + async createSession(filePath, _formatOptions, sessionId) { + creates.push(filePath) + assert.ok(sessionId) + await enter(`session:${sessionId}`) + existingSessions.add(sessionId) + return { success: true, sessionId } + }, + async appendSession(sessionId, filePath) { + appends.push({ sessionId, filePath }) + await enter(`session:${sessionId}`) + return { success: true, newMessageCount: 1 } + }, + } + + const results = await autoImportBatch( + [ + { filePath: 'group-a-1', sessionId: 'created-a' }, + { filePath: 'group-b', sessionId: 'created-b' }, + { filePath: 'group-a-2', sessionId: 'created-a' }, + { filePath: 'existing-1', sessionId: 'existing' }, + { filePath: 'existing-2', sessionId: 'existing' }, + ].map(({ filePath, sessionId }) => ({ + id: filePath, + filePath, + options: { explicitSessionId: sessionId }, + })), + deps, + { concurrency: 2 } + ) + + assert.equal(maxActive, 2) + assert.deepEqual(creates.sort(), ['group-a-1', 'group-b']) + assert.deepEqual( + appends.sort((left, right) => left.filePath.localeCompare(right.filePath)), + [ + { sessionId: 'existing', filePath: 'existing-1' }, + { sessionId: 'existing', filePath: 'existing-2' }, + { sessionId: 'created-a', filePath: 'group-a-2' }, + ] + ) + assert.deepEqual( + results.map((result) => result.status), + ['success', 'success', 'success', 'success', 'success'] + ) +}) + +test('re-resolves exclusive items after earlier writes to preserve serial matching semantics', async () => { + let createdSessionId: string | undefined + const creates: string[] = [] + const appends: string[] = [] + const deps: AutoImportDeps = { + listSessionIds: () => (createdSessionId ? [createdSessionId] : []), + openReadonly: () => { + throw new Error('matcher is stubbed') + }, + sessionExists: (sessionId) => sessionId === createdSessionId, + resolveTarget: async () => + createdSessionId + ? { action: 'incremental', sessionId: createdSessionId, matchedBy: 'trailing-messages' } + : { action: 'create', reason: 'no-match' }, + async createSession(filePath) { + creates.push(filePath) + createdSessionId = 'created' + return { success: true, sessionId: createdSessionId } + }, + async appendSession(sessionId, filePath) { + assert.equal(sessionId, 'created') + appends.push(filePath) + return { success: true, newMessageCount: 1 } + }, + } + const exclusivePlan: AutoImportTargetPlan = { + decision: { action: 'create', reason: 'no-match' }, + concurrencyKey: 'unresolved', + exclusive: true, + coalesceCreate: false, + } + const staleStablePlan: AutoImportTargetPlan = { + decision: { action: 'create', reason: 'no-match' }, + concurrencyKey: 'source:second', + exclusive: false, + coalesceCreate: true, + } + + const results = await autoImportBatch( + [ + { id: 'first', filePath: 'first' }, + { id: 'second', filePath: 'second' }, + ], + deps, + { + concurrency: 2, + resolveTargetPlan: async (filePath) => (filePath === 'first' ? exclusivePlan : staleStablePlan), + } + ) + + assert.deepEqual(creates, ['first']) + assert.deepEqual(appends, ['second']) + assert.deepEqual( + results.map((result) => result.status), + ['success', 'success'] + ) +}) + +test('re-resolves a later stable identity after an earlier automatic import writes', async () => { + let createdSessionId: string | undefined + const creates: string[] = [] + const appends: Array<{ sessionId: string; filePath: string }> = [] + const deps: AutoImportDeps = { + listSessionIds: () => (createdSessionId ? [createdSessionId] : []), + openReadonly: () => { + throw new Error('preflight is stubbed') + }, + sessionExists: (sessionId) => sessionId === createdSessionId, + resolveTarget: async () => + createdSessionId + ? { action: 'incremental', sessionId: createdSessionId, matchedBy: 'trailing-messages' } + : { action: 'create', reason: 'no-match' }, + async createSession(filePath) { + creates.push(filePath) + createdSessionId = 'created' + return { success: true, sessionId: createdSessionId } + }, + async appendSession(sessionId, filePath) { + appends.push({ sessionId, filePath }) + return { success: true, newMessageCount: 1 } + }, + } + const plans = new Map([ + [ + 'before-drift', + { + decision: { action: 'create', reason: 'no-match' }, + concurrencyKey: 'source:old-stable-id', + exclusive: false, + coalesceCreate: true, + }, + ], + [ + 'after-drift', + { + decision: { action: 'create', reason: 'no-match' }, + concurrencyKey: 'source:new-stable-id', + exclusive: false, + coalesceCreate: true, + }, + ], + ]) + + const results = await autoImportBatch( + [ + { id: 'before', filePath: 'before-drift' }, + { id: 'after', filePath: 'after-drift' }, + ], + deps, + { + concurrency: 2, + resolveTargetPlan: async (filePath) => plans.get(filePath)!, + } + ) + + assert.deepEqual(creates, ['before-drift']) + assert.deepEqual(appends, [{ sessionId: 'created', filePath: 'after-drift' }]) + assert.deepEqual( + results.map((result) => result.status), + ['success', 'success'] + ) +}) + +test('re-resolves every trailing-message match after an earlier append changes the target tail', async () => { + let firstAppendFinished = false + const appends: string[] = [] + const creates: string[] = [] + const deps: AutoImportDeps = { + listSessionIds: () => ['existing'], + openReadonly: () => { + throw new Error('preflight is stubbed') + }, + sessionExists: (sessionId) => sessionId === 'existing', + resolveTarget: async () => + firstAppendFinished + ? { action: 'create', reason: 'no-match' } + : { action: 'incremental', sessionId: 'existing', matchedBy: 'trailing-messages' }, + async createSession(filePath) { + creates.push(filePath) + return { success: true, sessionId: 'created' } + }, + async appendSession(_sessionId, filePath) { + appends.push(filePath) + firstAppendFinished = true + return { success: true, newMessageCount: 1 } + }, + } + const staleTrailingPlan: AutoImportTargetPlan = { + decision: { action: 'incremental', sessionId: 'existing', matchedBy: 'trailing-messages' }, + concurrencyKey: 'unresolved', + exclusive: true, + coalesceCreate: false, + } + + const results = await autoImportBatch( + [ + { id: 'first', filePath: 'first' }, + { id: 'second', filePath: 'second' }, + ], + deps, + { + concurrency: 2, + resolveTargetPlan: async () => staleTrailingPlan, + } + ) + + assert.deepEqual(appends, ['first']) + assert.deepEqual(creates, ['second']) + assert.deepEqual( + results.map((result) => result.status), + ['success', 'success'] + ) +}) + +test('invalidates a later auto-resolved incremental plan after exclusive work writes', async () => { + let createdSessionId: string | undefined + const appends: Array<{ sessionId: string; filePath: string }> = [] + const deps: AutoImportDeps = { + listSessionIds: () => ['old', ...(createdSessionId ? [createdSessionId] : [])], + openReadonly: () => { + throw new Error('preflight is stubbed') + }, + sessionExists: (sessionId) => sessionId === 'old' || sessionId === createdSessionId, + resolveTarget: async () => + createdSessionId + ? { action: 'incremental', sessionId: createdSessionId, matchedBy: 'trailing-messages' } + : { action: 'create', reason: 'no-match' }, + async createSession() { + createdSessionId = 'created' + return { success: true, sessionId: createdSessionId } + }, + async appendSession(sessionId, filePath) { + appends.push({ sessionId, filePath }) + return { success: true, newMessageCount: 1 } + }, + } + const plans = new Map([ + [ + 'first', + { + decision: { action: 'create', reason: 'no-match' }, + concurrencyKey: 'unresolved', + exclusive: true, + coalesceCreate: false, + }, + ], + [ + 'second', + { + decision: { action: 'incremental', sessionId: 'old', matchedBy: 'stable-id' }, + concurrencyKey: 'session:old', + exclusive: false, + coalesceCreate: false, + }, + ], + ]) + + await autoImportBatch( + [ + { id: 'first', filePath: 'first' }, + { id: 'second', filePath: 'second' }, + ], + deps, + { + concurrency: 2, + resolveTargetPlan: async (filePath) => plans.get(filePath)!, + } + ) + + assert.deepEqual(appends, [{ sessionId: 'created', filePath: 'second' }]) +}) diff --git a/packages/node-runtime/src/import/auto-import-batch.ts b/packages/node-runtime/src/import/auto-import-batch.ts new file mode 100644 index 00000000..cbd4d234 --- /dev/null +++ b/packages/node-runtime/src/import/auto-import-batch.ts @@ -0,0 +1,187 @@ +import { resolveAutoImportTargetPlan, type AutoImportDecision, type AutoImportTargetPlan } from './auto-import-matcher' +import { autoImportFile, type AutoImportDeps, type AutoImportOptions, type AutoImportResult } from './auto-importer' +import { normalizeBatchConcurrency, runKeyedBatch, type KeyedBatchTaskResult } from './batch-coordinator' +import type { ImportProgressCallback } from './streaming-importer' +import { appLogger } from '../logging/app-logger' + +export interface AutoImportBatchItem { + id: string + filePath: string + options?: AutoImportOptions +} + +export type AutoImportBatchItemResult = + | { id: string; status: 'success'; result: AutoImportResult } + | { id: string; status: 'failed'; error: string; result?: AutoImportResult } + | { id: string; status: 'cancelled' } + +export interface AutoImportBatchOptions { + concurrency?: number + signal?: AbortSignal + onItemStart?: (item: AutoImportBatchItem, index: number) => void + onItemProgress?: (item: AutoImportBatchItem, index: number, progress: Parameters[0]) => void + onItemComplete?: (item: AutoImportBatchItem, index: number, result: AutoImportBatchItemResult) => void + resolveTargetPlan?: typeof resolveAutoImportTargetPlan +} + +interface PlannedItem { + item: AutoImportBatchItem + itemIndex: number + plan: AutoImportTargetPlan +} + +function mapTaskResult( + item: AutoImportBatchItem, + result: KeyedBatchTaskResult +): AutoImportBatchItemResult { + if (result.status === 'cancelled') return { id: item.id, status: 'cancelled' } + if (result.status === 'failed') return { id: item.id, status: 'failed', error: result.error } + if (!result.value.success) { + return { + id: item.id, + status: 'failed', + error: result.value.error ?? 'Import failed', + result: result.value, + } + } + return { id: item.id, status: 'success', result: result.value } +} + +function preserveSerialTargetMatching(planned: PlannedItem[]): void { + let priorPlannedWrite = false + + for (const plannedItem of planned) { + let plan = plannedItem.plan + if (priorPlannedWrite && !plannedItem.item.options?.explicitSessionId && !plan.exclusive) { + plan = { + ...plan, + concurrencyKey: `unresolved:${plannedItem.itemIndex}`, + exclusive: true, + coalesceCreate: false, + } + } + + plannedItem.plan = plan + priorPlannedWrite = true + } +} + +export async function autoImportBatch( + items: AutoImportBatchItem[], + deps: AutoImportDeps, + options: AutoImportBatchOptions = {} +): Promise { + if (items.length === 0) return [] + + const resolvePlan = options.resolveTargetPlan ?? resolveAutoImportTargetPlan + const concurrency = normalizeBatchConcurrency(options.concurrency, items.length) + appLogger.info('import-batch', 'Batch import started', { + itemCount: items.length, + concurrency, + }) + const preflight = await runKeyedBatch( + items.map((item, index) => ({ value: { item, itemIndex: index }, key: `preflight:${index}` })), + { + concurrency, + signal: options.signal, + run: async ({ item, itemIndex }) => { + const itemOptions = item.options ?? {} + if (itemOptions.explicitSessionId) { + const sessionId = itemOptions.explicitSessionId + const decision: AutoImportDecision = deps.sessionExists(sessionId) + ? { action: 'incremental', sessionId, matchedBy: 'source-session-id' } + : { action: 'create', reason: 'no-match' } + return { + item, + itemIndex, + plan: { + decision, + concurrencyKey: `session:${sessionId}`, + exclusive: false, + coalesceCreate: decision.action === 'create', + }, + } satisfies PlannedItem + } + return { + item, + itemIndex, + plan: await resolvePlan( + item.filePath, + { + ...deps, + onProgress: (progress) => options.onItemProgress?.(item, itemIndex, progress), + }, + itemOptions.formatOptions + ), + } satisfies PlannedItem + }, + } + ) + + const finalResults: Array = Array(items.length) + const planned: PlannedItem[] = [] + for (let index = 0; index < preflight.length; index++) { + const result = preflight[index] + if (result.status === 'success') { + planned.push(result.value) + continue + } + const mapped: AutoImportBatchItemResult = + result.status === 'cancelled' + ? { id: items[index].id, status: 'cancelled' } + : { id: items[index].id, status: 'failed', error: result.error } + finalResults[index] = mapped + options.onItemComplete?.(items[index], index, mapped) + } + + preserveSerialTargetMatching(planned) + const targetSessionByKey = new Map() + await runKeyedBatch( + planned.map((plannedItem) => ({ + value: plannedItem, + key: plannedItem.plan.concurrencyKey, + exclusive: plannedItem.plan.exclusive, + })), + { + concurrency, + signal: options.signal, + onTaskStart: ({ item, itemIndex }) => options.onItemStart?.(item, itemIndex), + onTaskComplete: ({ item, itemIndex }, _plannedIndex, taskResult) => { + const mapped = mapTaskResult(item, taskResult) + finalResults[itemIndex] = mapped + options.onItemComplete?.(item, itemIndex, mapped) + }, + run: async ({ item, itemIndex, plan }) => { + const targetSessionId = plan.coalesceCreate ? targetSessionByKey.get(plan.concurrencyKey) : undefined + const decision: AutoImportDecision = targetSessionId + ? { action: 'incremental', sessionId: targetSessionId, matchedBy: 'trailing-messages' } + : plan.decision + const result = await autoImportFile( + item.filePath, + { + ...deps, + onProgress: (progress) => options.onItemProgress?.(item, itemIndex, progress), + }, + { + ...item.options, + resolvedDecision: + plan.exclusive || (item.options?.explicitSessionId && !targetSessionId) ? undefined : decision, + } + ) + if (!plan.exclusive && result.success && result.sessionId) { + targetSessionByKey.set(plan.concurrencyKey, result.sessionId) + } + return result + }, + } + ) + + const completedResults = finalResults as AutoImportBatchItemResult[] + appLogger.info('import-batch', 'Batch import completed', { + itemCount: items.length, + successCount: completedResults.filter((result) => result.status === 'success').length, + failedCount: completedResults.filter((result) => result.status === 'failed').length, + cancelledCount: completedResults.filter((result) => result.status === 'cancelled').length, + }) + return completedResults +} diff --git a/packages/node-runtime/src/import/auto-import-matcher.test.ts b/packages/node-runtime/src/import/auto-import-matcher.test.ts index adf4db20..ed90be49 100644 --- a/packages/node-runtime/src/import/auto-import-matcher.test.ts +++ b/packages/node-runtime/src/import/auto-import-matcher.test.ts @@ -6,7 +6,7 @@ import test from 'node:test' import { CHAT_DB_SCHEMA } from '@openchatlab/core' import { MessageType, type ParsedMessage } from '@openchatlab/shared-types' import { openBetterSqliteDatabase } from '../better-sqlite3-adapter' -import { resolveAutoImportTarget, type AutoImportMatcherDeps } from './auto-import-matcher' +import { resolveAutoImportTarget, resolveAutoImportTargetPlan, type AutoImportMatcherDeps } from './auto-import-matcher' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -98,6 +98,47 @@ function textMessage(sender: string, timestamp: number, content: string, type = } } +test('derives a hashed parallel key only from a stable source identity', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl( + sourcePath, + { name: 'New group', platform: 'qq', type: 'group', groupId: 'private-group-id' }, + [{ platformId: 'user-1', accountName: 'Alice' }], + [textMessage('user-1', 1783840001, 'hello')] + ) + + const plan = await resolveAutoImportTargetPlan(sourcePath, createDeps(tempDir, [])) + + assert.deepEqual(plan.decision, { action: 'create', reason: 'no-match' }) + assert.match(plan.concurrencyKey, /^source:[a-f0-9]{64}$/) + assert.doesNotMatch(plan.concurrencyKey, /private-group-id/) + assert.equal(plan.exclusive, false) + assert.equal(plan.coalesceCreate, true) +}) + +test('makes sources without a stable identity exclusive', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl( + sourcePath, + { name: 'Unknown private chat', platform: 'line', type: 'private' }, + [{ platformId: 'Alice', accountName: 'Alice' }], + [textMessage('Alice', 1783840001, 'hello')] + ) + + const plan = await resolveAutoImportTargetPlan(sourcePath, createDeps(tempDir, [])) + + assert.deepEqual(plan, { + decision: { action: 'create', reason: 'no-match' }, + concurrencyKey: 'unresolved', + exclusive: true, + coalesceCreate: false, + }) +}) + test('matches a unique group session by stable group id', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) diff --git a/packages/node-runtime/src/import/auto-import-matcher.ts b/packages/node-runtime/src/import/auto-import-matcher.ts index e380e2ea..70fed819 100644 --- a/packages/node-runtime/src/import/auto-import-matcher.ts +++ b/packages/node-runtime/src/import/auto-import-matcher.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto' import { generateMessageKey, getSessionMeta, isChatSessionDb, type DatabaseAdapter } from '@openchatlab/core' import { streamParseFile, type ParsedMeta } from '@openchatlab/parser' import { MessageType } from '@openchatlab/shared-types' @@ -13,6 +14,13 @@ export type AutoImportDecision = | { action: 'incremental'; sessionId: string; matchedBy: AutoImportMatchMethod } | { action: 'create'; reason: AutoImportCreateReason } +export interface AutoImportTargetPlan { + decision: AutoImportDecision + concurrencyKey: string + exclusive: boolean + coalesceCreate: boolean +} + export interface AutoImportMatcherDeps { listSessionIds(): string[] openReadonly(sessionId: string): DatabaseAdapter @@ -48,6 +56,14 @@ export async function resolveAutoImportTarget( deps: AutoImportMatcherDeps, formatOptions?: Record ): Promise { + return (await resolveAutoImportTargetPlan(filePath, deps, formatOptions)).decision +} + +export async function resolveAutoImportTargetPlan( + filePath: string, + deps: AutoImportMatcherDeps, + formatOptions?: Record +): Promise { let sourceMeta: ParsedMeta | null = null const sourceMemberIds = new Set() const sourceWindows = new Set() @@ -80,6 +96,14 @@ export async function resolveAutoImportTarget( const meta = sourceMeta as ParsedMeta const privateIdentity = meta.type === 'private' ? buildPrivateIdentity(meta.ownerId, sourceMemberIds) : null const hasStableIdentity = Boolean(meta.groupId || privateIdentity) + const stableIdentity = meta.groupId + ? `group\0${meta.groupId}` + : privateIdentity + ? `private\0${privateIdentity}` + : null + const stableConcurrencyKey = stableIdentity + ? `source:${createHash('sha256').update(`${meta.platform}\0${meta.type}\0${stableIdentity}`).digest('hex')}` + : null const stableMatches: string[] = [] const trailingMatches: string[] = [] @@ -148,20 +172,54 @@ export async function resolveAutoImportTarget( if (hasStableIdentity) { if (meta.sourceSessionId && stableMatches.includes(meta.sourceSessionId)) { - return { action: 'incremental', sessionId: meta.sourceSessionId, matchedBy: 'source-session-id' } + return { + decision: { action: 'incremental', sessionId: meta.sourceSessionId, matchedBy: 'source-session-id' }, + concurrencyKey: `session:${meta.sourceSessionId}`, + exclusive: false, + coalesceCreate: false, + } } if (stableMatches.length === 1) { - return { action: 'incremental', sessionId: stableMatches[0], matchedBy: 'stable-id' } + return { + decision: { action: 'incremental', sessionId: stableMatches[0], matchedBy: 'stable-id' }, + concurrencyKey: `session:${stableMatches[0]}`, + exclusive: false, + coalesceCreate: false, + } } } // 稳定身份缺失、漂移或产生多个候选时，仍只接受唯一的 5 条连续消息重叠。 if (meta.sourceSessionId && trailingMatches.includes(meta.sourceSessionId)) { - return { action: 'incremental', sessionId: meta.sourceSessionId, matchedBy: 'source-session-id' } + return { + decision: { action: 'incremental', sessionId: meta.sourceSessionId, matchedBy: 'source-session-id' }, + concurrencyKey: `session:${meta.sourceSessionId}`, + exclusive: false, + coalesceCreate: false, + } } if (trailingMatches.length === 1) { - return { action: 'incremental', sessionId: trailingMatches[0], matchedBy: 'trailing-messages' } + return { + decision: { action: 'incremental', sessionId: trailingMatches[0], matchedBy: 'trailing-messages' }, + concurrencyKey: `session:${trailingMatches[0]}`, + exclusive: false, + coalesceCreate: false, + } } const ambiguous = stableMatches.length > 1 || trailingMatches.length > 1 - return { action: 'create', reason: ambiguous ? 'ambiguous' : 'no-match' } + const decision: AutoImportDecision = { action: 'create', reason: ambiguous ? 'ambiguous' : 'no-match' } + if (!ambiguous && stableConcurrencyKey) { + return { + decision, + concurrencyKey: stableConcurrencyKey, + exclusive: false, + coalesceCreate: true, + } + } + return { + decision, + concurrencyKey: 'unresolved', + exclusive: true, + coalesceCreate: false, + } } diff --git a/packages/node-runtime/src/import/auto-importer.ts b/packages/node-runtime/src/import/auto-importer.ts index 017a4fa9..cf47b2f4 100644 --- a/packages/node-runtime/src/import/auto-importer.ts +++ b/packages/node-runtime/src/import/auto-importer.ts @@ -14,6 +14,8 @@ import { isValidImportSessionId } from './session-id' export interface AutoImportOptions { explicitSessionId?: string formatOptions?: Record + /** Pre-resolved by the batch coordinator while the data directory is read-only. */ + resolvedDecision?: AutoImportDecision } export interface AutoImportDeps extends AutoImportMatcherDeps { @@ -83,6 +85,13 @@ async function planAutoImport( deps: Pick, options: AutoImportOptions ): Promise { + if (options.resolvedDecision) { + const decision = options.resolvedDecision + return decision.action === 'incremental' + ? { action: 'incremental', sessionId: decision.sessionId, matchedBy: decision.matchedBy } + : { action: 'create', reason: decision.reason } + } + if (options.explicitSessionId) { if (!isValidImportSessionId(options.explicitSessionId)) { throw new Error('sessionId contains invalid characters') diff --git a/packages/node-runtime/src/import/batch-coordinator.test.ts b/packages/node-runtime/src/import/batch-coordinator.test.ts new file mode 100644 index 00000000..1069ef48 --- /dev/null +++ b/packages/node-runtime/src/import/batch-coordinator.test.ts @@ -0,0 +1,132 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { setTimeout as delay } from 'node:timers/promises' +import { resolveDefaultBatchConcurrency, runKeyedBatch } from './batch-coordinator' + +test('uses two workers by default and degrades on low-resource runtimes', () => { + assert.equal(resolveDefaultBatchConcurrency(100, { cpuCount: 8, totalMemoryBytes: 16 * 1024 ** 3 }), 2) + assert.equal(resolveDefaultBatchConcurrency(100, { cpuCount: 2, totalMemoryBytes: 16 * 1024 ** 3 }), 1) + assert.equal(resolveDefaultBatchConcurrency(100, { cpuCount: 8, totalMemoryBytes: 2 * 1024 ** 3 }), 1) + assert.equal(resolveDefaultBatchConcurrency(1, { cpuCount: 8, totalMemoryBytes: 16 * 1024 ** 3 }), 1) +}) + +test('runs different keys concurrently and serializes identical keys', async () => { + let active = 0 + let maxActive = 0 + const activeKeys = new Set() + const tasks = [ + { value: 'a-1', key: 'a' }, + { value: 'b-1', key: 'b' }, + { value: 'a-2', key: 'a' }, + { value: 'c-1', key: 'c' }, + ] + + const results = await runKeyedBatch(tasks, { + concurrency: 2, + async run(value, index) { + const key = tasks[index].key + assert.equal(activeKeys.has(key), false) + activeKeys.add(key) + active++ + maxActive = Math.max(maxActive, active) + await delay(5) + active-- + activeKeys.delete(key) + return value + }, + }) + + assert.equal(maxActive, 2) + assert.deepEqual( + results.map((result) => (result.status === 'success' ? result.value : result.status)), + ['a-1', 'b-1', 'a-2', 'c-1'] + ) +}) + +test('treats exclusive work as an ordering barrier', async () => { + const events: string[] = [] + const tasks = [ + { value: 'before', key: 'before' }, + { value: 'exclusive', key: 'exclusive', exclusive: true }, + { value: 'after', key: 'after' }, + ] + + await runKeyedBatch(tasks, { + concurrency: 3, + async run(value) { + events.push(`start:${value}`) + await delay(2) + events.push(`end:${value}`) + return value + }, + }) + + assert.deepEqual(events, [ + 'start:before', + 'end:before', + 'start:exclusive', + 'end:exclusive', + 'start:after', + 'end:after', + ]) +}) + +test('does not start pending work after cancellation and keeps completed results', async () => { + const controller = new AbortController() + let releaseFirst!: () => void + const firstRunning = new Promise((resolve) => { + releaseFirst = resolve + }) + const started: string[] = [] + + const resultPromise = runKeyedBatch( + [ + { value: 'first', key: 'first' }, + { value: 'second', key: 'second' }, + { value: 'third', key: 'third' }, + ], + { + concurrency: 1, + signal: controller.signal, + async run(value) { + started.push(value) + if (value === 'first') await firstRunning + return value + }, + } + ) + + await delay(0) + controller.abort() + releaseFirst() + const results = await resultPromise + + assert.deepEqual(started, ['first']) + assert.deepEqual( + results.map((result) => result.status), + ['success', 'cancelled', 'cancelled'] + ) +}) + +test('aggregates failures without stopping independent work', async () => { + const results = await runKeyedBatch( + [ + { value: 'ok-1', key: 'a' }, + { value: 'bad', key: 'b' }, + { value: 'ok-2', key: 'c' }, + ], + { + concurrency: 2, + async run(value) { + if (value === 'bad') throw new Error('expected failure') + return value + }, + } + ) + + assert.deepEqual(results, [ + { status: 'success', value: 'ok-1' }, + { status: 'failed', error: 'expected failure' }, + { status: 'success', value: 'ok-2' }, + ]) +}) diff --git a/packages/node-runtime/src/import/batch-coordinator.ts b/packages/node-runtime/src/import/batch-coordinator.ts new file mode 100644 index 00000000..0d0a1826 --- /dev/null +++ b/packages/node-runtime/src/import/batch-coordinator.ts @@ -0,0 +1,128 @@ +export interface KeyedBatchTask { + value: T + key: string + /** + * Exclusive tasks form a barrier: they start only after all earlier work + * finishes, and no later work starts until they complete. + */ + exclusive?: boolean +} + +export type KeyedBatchTaskResult = + | { status: 'success'; value: T } + | { status: 'failed'; error: string } + | { status: 'cancelled' } + +export interface KeyedBatchOptions { + concurrency?: number + signal?: AbortSignal + run(task: TInput, index: number): Promise + onTaskStart?: (task: TInput, index: number) => void + onTaskComplete?: (task: TInput, index: number, result: KeyedBatchTaskResult) => void +} + +export function normalizeBatchConcurrency(value: unknown, taskCount: number): number { + if (taskCount <= 0) return 0 + if (typeof value !== 'number' || !Number.isInteger(value) || value < 1) return 1 + return Math.min(value, taskCount) +} + +export function resolveDefaultBatchConcurrency( + taskCount: number, + resources: { cpuCount?: number; totalMemoryBytes?: number } = {} +): number { + if (taskCount <= 0) return 0 + const cpuCount = resources.cpuCount ?? cpus().length + const totalMemoryBytes = resources.totalMemoryBytes ?? totalmem() + const lowResourceRuntime = cpuCount < 4 || totalMemoryBytes < 4 * 1024 * 1024 * 1024 + return Math.min(taskCount, lowResourceRuntime ? 1 : 2) +} + +/** + * Run keyed work with bounded concurrency while serializing identical keys. + * Exclusive tasks are ordering barriers for work whose target cannot be + * resolved safely before writes begin. + */ +export async function runKeyedBatch( + tasks: Array>, + options: KeyedBatchOptions +): Promise>> { + if (tasks.length === 0) return [] + + const concurrency = normalizeBatchConcurrency(options.concurrency, tasks.length) + const results: Array | undefined> = Array(tasks.length) + const pending = new Set(tasks.map((_, index) => index)) + const activeKeys = new Set() + let activeCount = 0 + let exclusiveActive = false + + return new Promise((resolve) => { + const finishIfDone = () => { + if (activeCount !== 0 || pending.size !== 0) return false + options.signal?.removeEventListener('abort', schedule) + resolve(results as Array>) + return true + } + + const cancelPending = () => { + if (!options.signal?.aborted) return + for (const index of pending) { + pending.delete(index) + const result = { status: 'cancelled' as const } + results[index] = result + options.onTaskComplete?.(tasks[index].value, index, result) + } + } + + const findRunnableIndex = (): number | undefined => { + if (exclusiveActive) return undefined + for (const index of pending) { + const task = tasks[index] + if (task.exclusive) return activeCount === 0 ? index : undefined + if (!activeKeys.has(task.key)) return index + } + return undefined + } + + function schedule(): void { + cancelPending() + if (finishIfDone()) return + + while (activeCount < concurrency && !options.signal?.aborted) { + const index = findRunnableIndex() + if (index === undefined) break + + const task = tasks[index] + pending.delete(index) + activeCount++ + activeKeys.add(task.key) + if (task.exclusive) exclusiveActive = true + options.onTaskStart?.(task.value, index) + + void Promise.resolve() + .then(() => options.run(task.value, index)) + .then( + (value) => ({ status: 'success' as const, value }), + (error) => ({ + status: 'failed' as const, + error: error instanceof Error ? error.message : String(error), + }) + ) + .then((result) => { + results[index] = result + options.onTaskComplete?.(task.value, index, result) + }) + .finally(() => { + activeCount-- + activeKeys.delete(task.key) + if (task.exclusive) exclusiveActive = false + schedule() + }) + } + } + + options.signal?.addEventListener('abort', schedule, { once: true }) + schedule() + }) +} +import { cpus, totalmem } from 'node:os' diff --git a/packages/node-runtime/src/import/index.ts b/packages/node-runtime/src/import/index.ts index 03d3c466..7ec7c651 100644 --- a/packages/node-runtime/src/import/index.ts +++ b/packages/node-runtime/src/import/index.ts @@ -3,6 +3,7 @@ export type { ImportMeta, WriteParseResultStats } from './write-parse-result' export { logNativeParserStatus } from './native-parser-status' export { LogLevel, + createImportPerfLogger, initPerfLog, logPerf, logPerfDetail, @@ -13,6 +14,7 @@ export { getErrorCount, logSummary, } from './perf-logger' +export type { ImportPerfLogger } from './perf-logger' export { streamingImport, analyzeNewImport, streamParseFileInfo } from './streaming-importer' export { MessageBatchInserter, MESSAGE_INSERT_MAX_ROWS } from './message-batch-inserter' export type { @@ -39,12 +41,13 @@ export type { } from './incremental-importer' export { isValidImportSessionId } from './session-id' -export { resolveAutoImportTarget } from './auto-import-matcher' +export { resolveAutoImportTarget, resolveAutoImportTargetPlan } from './auto-import-matcher' export type { AutoImportCreateReason, AutoImportDecision, AutoImportMatcherDeps, AutoImportMatchMethod, + AutoImportTargetPlan, } from './auto-import-matcher' export { analyzeAutoImportFile, autoImportFile } from './auto-importer' export type { @@ -54,6 +57,10 @@ export type { AutoImportOptions, AutoImportResult, } from './auto-importer' +export { autoImportBatch } from './auto-import-batch' +export type { AutoImportBatchItem, AutoImportBatchItemResult, AutoImportBatchOptions } from './auto-import-batch' +export { normalizeBatchConcurrency, resolveDefaultBatchConcurrency, runKeyedBatch } from './batch-coordinator' +export type { KeyedBatchOptions, KeyedBatchTask, KeyedBatchTaskResult } from './batch-coordinator' export { IMPORT_IN_PROGRESS_ERROR_KEY, IMPORT_LOCK_FILENAME, diff --git a/packages/node-runtime/src/import/perf-logger.test.ts b/packages/node-runtime/src/import/perf-logger.test.ts new file mode 100644 index 00000000..317204a0 --- /dev/null +++ b/packages/node-runtime/src/import/perf-logger.test.ts @@ -0,0 +1,35 @@ +import assert from 'node:assert/strict' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import test from 'node:test' +import { createImportPerfLogger } from './perf-logger' + +test('keeps concurrent import performance logs isolated', (t) => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-perf-log-')) + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + + const first = createImportPerfLogger(root) + const second = createImportPerfLogger(root) + first.init('first') + second.init('second') + first.info('first-only') + second.info('second-only') + first.error('first-error') + + const firstPath = first.getCurrentLogFile() + const secondPath = second.getCurrentLogFile() + assert.ok(firstPath) + assert.ok(secondPath) + assert.notEqual(firstPath, secondPath) + assert.match(fs.readFileSync(firstPath, 'utf8'), /first-only/) + assert.doesNotMatch(fs.readFileSync(firstPath, 'utf8'), /second-only/) + assert.match(fs.readFileSync(secondPath, 'utf8'), /second-only/) + assert.doesNotMatch(fs.readFileSync(secondPath, 'utf8'), /first-only/) + assert.equal(first.getErrorCount(), 1) + assert.equal(second.getErrorCount(), 0) + + first.reset() + assert.equal(first.getCurrentLogFile(), null) + assert.equal(second.getCurrentLogFile(), secondPath) +}) diff --git a/packages/node-runtime/src/import/perf-logger.ts b/packages/node-runtime/src/import/perf-logger.ts index 8085a01c..e5f1e216 100644 --- a/packages/node-runtime/src/import/perf-logger.ts +++ b/packages/node-runtime/src/import/perf-logger.ts @@ -1,10 +1,8 @@ /** * Import performance logger. * - * Extracted from electron/main/worker/core/perfLogger.ts. - * Records real-time performance metrics during import operations. - * - * Caller provides the log directory; this module has no path assumptions. + * Each import must own an independent logger instance because batch imports can + * run concurrently in the same process. */ import * as fs from 'fs' @@ -15,117 +13,139 @@ export enum LogLevel { INFO = 'INFO', } -let lastLogTime = Date.now() -let lastMessageCount = 0 -let currentLogFile: string | null = null -let errorCount = 0 - -export function initPerfLog(sessionId: string, logDir: string): void { - try { - if (!fs.existsSync(logDir)) { - fs.mkdirSync(logDir, { recursive: true }) - } - currentLogFile = path.join(logDir, `import_${sessionId}_${Date.now()}.log`) - fs.writeFileSync(currentLogFile, `=== Import Log ===\nStart time: ${new Date().toISOString()}\n\n`, 'utf-8') - } catch { - // Ignore initialization failure - } +export interface ImportPerfLogger { + info(message: string): void + error(message: string, error?: Error): void + perf(event: string, messagesProcessed: number, batchSize?: number): void + perfDetail(detail: string): void + summary(totalMessages: number, totalMembers: number): void + reset(): void + init(sessionId: string): void + getCurrentLogFile(): string | null + getErrorCount(): number } -export function logPerf(event: string, messagesProcessed: number, batchSize?: number): void { - const now = Date.now() - const duration = now - lastLogTime - const messagesDelta = messagesProcessed - lastMessageCount - const speed = duration > 0 ? Math.round((messagesDelta / duration) * 1000) : 0 +export function createImportPerfLogger(logDir: string): ImportPerfLogger { + let lastLogTime = Date.now() + let lastMessageCount = 0 + let currentLogFile: string | null = null + let errorCount = 0 - let memory = 0 - try { - const used = process.memoryUsage() - memory = Math.round(used.heapUsed / 1024 / 1024) - } catch { - // Ignore - } - - const logLine = - `[${new Date().toISOString()}] ${event} | ` + - `messages: ${messagesProcessed.toLocaleString()} | ` + - `elapsed: ${duration}ms | ` + - `speed: ${speed.toLocaleString()}/s | ` + - `memory: ${memory}MB` + - (batchSize ? ` | batch: ${batchSize}` : '') + - '\n' - - if (currentLogFile) { + const append = (line: string) => { + if (!currentLogFile) return try { - fs.appendFileSync(currentLogFile, logLine, 'utf-8') + fs.appendFileSync(currentLogFile, line, 'utf-8') } catch { - // Ignore write failure + // Ignore write failure. } } - lastLogTime = now - lastMessageCount = messagesProcessed -} - -export function logPerfDetail(detail: string): void { - if (currentLogFile) { - try { - fs.appendFileSync(currentLogFile, ` ${detail}\n`, 'utf-8') - } catch { - // Ignore - } + const writeLogLine = (level: LogLevel, message: string) => { + append(`[${new Date().toISOString()}] [${level}] ${message}\n`) } -} -export function resetPerfLog(): void { - lastLogTime = Date.now() - lastMessageCount = 0 - currentLogFile = null - errorCount = 0 -} - -export function getCurrentLogFile(): string | null { - return currentLogFile -} - -function writeLogLine(level: LogLevel, message: string): void { - if (!currentLogFile) return - - const logLine = `[${new Date().toISOString()}] [${level}] ${message}\n` - try { - fs.appendFileSync(currentLogFile, logLine, 'utf-8') - } catch { - // Ignore write failure - } -} - -export function logError(message: string, error?: Error): void { - errorCount++ - const errorDetail = error ? `: ${error.message}` : '' - writeLogLine(LogLevel.ERROR, `${message}${errorDetail}`) -} - -export function logInfo(message: string): void { - writeLogLine(LogLevel.INFO, message) -} - -export function getErrorCount(): number { - return errorCount -} - -export function logSummary(totalMessages: number, totalMembers: number): void { - if (!currentLogFile) return - - const summary = ` + return { + init(sessionId) { + try { + if (!fs.existsSync(logDir)) fs.mkdirSync(logDir, { recursive: true }) + currentLogFile = path.join(logDir, `import_${sessionId}_${Date.now()}.log`) + fs.writeFileSync(currentLogFile, `=== Import Log ===\nStart time: ${new Date().toISOString()}\n\n`, 'utf-8') + } catch { + currentLogFile = null + } + }, + perf(event, messagesProcessed, batchSize) { + const now = Date.now() + const duration = now - lastLogTime + const messagesDelta = messagesProcessed - lastMessageCount + const speed = duration > 0 ? Math.round((messagesDelta / duration) * 1000) : 0 + let memory = 0 + try { + memory = Math.round(process.memoryUsage().heapUsed / 1024 / 1024) + } catch { + // Ignore memory sampling failure. + } + append( + `[${new Date().toISOString()}] ${event} | ` + + `messages: ${messagesProcessed.toLocaleString()} | ` + + `elapsed: ${duration}ms | ` + + `speed: ${speed.toLocaleString()}/s | ` + + `memory: ${memory}MB` + + (batchSize ? ` | batch: ${batchSize}` : '') + + '\n' + ) + lastLogTime = now + lastMessageCount = messagesProcessed + }, + perfDetail(detail) { + append(` ${detail}\n`) + }, + reset() { + lastLogTime = Date.now() + lastMessageCount = 0 + currentLogFile = null + errorCount = 0 + }, + getCurrentLogFile() { + return currentLogFile + }, + error(message, error) { + errorCount++ + writeLogLine(LogLevel.ERROR, `${message}${error ? `: ${error.message}` : ''}`) + }, + info(message) { + writeLogLine(LogLevel.INFO, message) + }, + getErrorCount() { + return errorCount + }, + summary(totalMessages, totalMembers) { + append(` === Import Summary === End time: ${new Date().toISOString()} Total messages: ${totalMessages.toLocaleString()} Total members: ${totalMembers.toLocaleString()} Errors: ${errorCount} -` - try { - fs.appendFileSync(currentLogFile, summary, 'utf-8') - } catch { - // Ignore +`) + }, } } + +let defaultLogger = createImportPerfLogger('') + +export function initPerfLog(sessionId: string, logDir: string): void { + defaultLogger = createImportPerfLogger(logDir) + defaultLogger.init(sessionId) +} + +export function logPerf(event: string, messagesProcessed: number, batchSize?: number): void { + defaultLogger.perf(event, messagesProcessed, batchSize) +} + +export function logPerfDetail(detail: string): void { + defaultLogger.perfDetail(detail) +} + +export function resetPerfLog(): void { + defaultLogger.reset() +} + +export function getCurrentLogFile(): string | null { + return defaultLogger.getCurrentLogFile() +} + +export function logError(message: string, error?: Error): void { + defaultLogger.error(message, error) +} + +export function logInfo(message: string): void { + defaultLogger.info(message) +} + +export function getErrorCount(): number { + return defaultLogger.getErrorCount() +} + +export function logSummary(totalMessages: number, totalMembers: number): void { + defaultLogger.summary(totalMessages, totalMembers) +} diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 6a860b6c..e71529f7 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -25,8 +25,13 @@ export { analyzeIncrementalImport, incrementalImport, resolveAutoImportTarget, + resolveAutoImportTargetPlan, analyzeAutoImportFile, autoImportFile, + autoImportBatch, + normalizeBatchConcurrency, + resolveDefaultBatchConcurrency, + runKeyedBatch, isValidImportSessionId, IMPORT_IN_PROGRESS_ERROR_KEY, IMPORT_LOCK_FILENAME, @@ -63,11 +68,18 @@ export type { AutoImportCreateReason, AutoImportMatcherDeps, AutoImportMatchMethod, + AutoImportTargetPlan, AutoImportAnalysisDeps, AutoImportAnalysisResult, AutoImportDeps, AutoImportOptions, AutoImportResult, + AutoImportBatchItem, + AutoImportBatchItemResult, + AutoImportBatchOptions, + KeyedBatchOptions, + KeyedBatchTask, + KeyedBatchTaskResult, ArchiveEntrySummary, ArchiveEntryStreamOpener, ArchiveEntryVisitor, @@ -80,6 +92,7 @@ export type { } from './import' export { LogLevel, + createImportPerfLogger, initPerfLog, logPerf, logPerfDetail, diff --git a/scripts/bench-streaming-import.mts b/scripts/bench-streaming-import.mts index d848466a..398a50a4 100644 --- a/scripts/bench-streaming-import.mts +++ b/scripts/bench-streaming-import.mts @@ -6,7 +6,7 @@ * * Usage: * pnpm exec tsx scripts/bench-streaming-import.mts single 100000 [runs=3] - * pnpm exec tsx scripts/bench-streaming-import.mts batch 100 10000 [runs=3] + * pnpm exec tsx scripts/bench-streaming-import.mts batch 100 10000 [runs=3] [concurrency=1] */ import { createHash } from 'node:crypto' @@ -20,6 +20,7 @@ import { CHAT_DB_TABLES } from '@openchatlab/core' import { getNativeParserStatus, isNativeFormatAvailable, type NativeParserStatus } from '@openchatlab/parser' import { BetterSqliteAdapter } from '../packages/node-runtime/src/better-sqlite3-adapter' import { computeAndSetOverviewCache } from '../packages/node-runtime/src/cache/session-cache' +import { runKeyedBatch } from '../packages/node-runtime/src/import/batch-coordinator' import { streamingImport, type ImportLogger, @@ -50,6 +51,7 @@ interface BenchmarkResult { scenario: 'single' | 'batch' parser: BenchmarkParser fileCount: number + concurrency: number inputMessages: number inputBytes: number durationMs: number @@ -280,7 +282,7 @@ export function inspectDatabase(dbPath: string): { } } -async function runWorker(fileCount: number, messagesPerFile: number): Promise { +async function runWorker(fileCount: number, messagesPerFile: number, concurrency: number): Promise { if (typeof global.gc !== 'function') { throw new Error('Benchmark worker requires Node.js --expose-gc') } @@ -311,23 +313,32 @@ async function runWorker(fileCount: number, messagesPerFile: number): Promise ({ + value: { inputPath, index }, + key: `session:${index}`, + })), + { + concurrency, + async run({ inputPath, index }) { + const dbPath = path.join(root, `session-${index}.db`) + dbPaths[index] = dbPath + const parserMonitor = createBenchmarkParserMonitor() + const result = await streamingImport( + inputPath, + createImportDeps(dbPath, cacheDir, parserMonitor.logger), + { formatId: 'weflow' }, + `bench-${index}` + ) + if (!result.success || !result.diagnostics?.performance) { + throw new Error(`Import ${index} failed: ${result.error ?? 'missing performance diagnostics'}`) + } + parserMonitor.assertRustNativeCompleted(index) + addTimings(timings, result.diagnostics.performance.timings) + sampledPeakRssMb = Math.max(sampledPeakRssMb, result.diagnostics.performance.rssSampledPeakMb) + }, } - parserMonitor.assertRustNativeCompleted(index) - addTimings(timings, result.diagnostics.performance.timings) - sampledPeakRssMb = Math.max(sampledPeakRssMb, result.diagnostics.performance.rssSampledPeakMb) - } + ) const durationMs = performance.now() - startedAt let messagesWritten = 0 @@ -346,6 +357,7 @@ async function runWorker(fileCount: number, messagesPerFile: number): Promise { - const [mode, first, second, third] = process.argv.slice(2) + const [mode, first, second, third, fourth] = process.argv.slice(2) if (mode === '--worker') { - const result = await runWorker(Number(first), Number(second)) + const result = await runWorker(Number(first), Number(second), Number(third)) console.log(`BENCH_RESULT ${JSON.stringify(result)}`) return } @@ -405,27 +425,30 @@ async function main(): Promise { let fileCount: number let messagesPerFile: number let runs: number + let concurrency: number if (mode === 'single') { fileCount = 1 messagesPerFile = Number(first ?? 100_000) runs = Number(second ?? 3) + concurrency = 1 } else if (mode === 'batch') { fileCount = Number(first ?? 100) messagesPerFile = Number(second ?? 10_000) runs = Number(third ?? 3) + concurrency = Number(fourth ?? 1) } else { - throw new Error('Usage: single [runs=3] | batch [runs=3]') + throw new Error('Usage: single [runs=3] | batch [runs=3] [concurrency=1]') } - if (![fileCount, messagesPerFile, runs].every((value) => Number.isInteger(value) && value > 0)) { + if (![fileCount, messagesPerFile, runs, concurrency].every((value) => Number.isInteger(value) && value > 0)) { throw new Error('All benchmark counts must be positive integers') } console.log( - `Benchmarking ${fileCount} file(s), ${messagesPerFile.toLocaleString()} messages/file, ${runs} isolated run(s)` + `Benchmarking ${fileCount} file(s), ${messagesPerFile.toLocaleString()} messages/file, concurrency ${concurrency}, ${runs} isolated run(s)` ) const results: BenchmarkResult[] = [] for (let index = 0; index < runs; index++) { - const result = runIsolated(fileCount, messagesPerFile) + const result = runIsolated(fileCount, messagesPerFile, concurrency) results.push(result) printResult(result, `run ${index + 1}`) } diff --git a/src/services/import/electron.ts b/src/services/import/electron.ts index d12ec1ed..5af508aa 100644 --- a/src/services/import/electron.ts +++ b/src/services/import/electron.ts @@ -14,6 +14,9 @@ import type { DemoImportResult, IncrementalAnalysis, IncrementalImportResult, + BatchImportItem, + BatchImportItemResult, + BatchImportProgress, } from './types' import { normalizeImportResult } from './types' @@ -23,6 +26,8 @@ function resolveFilePath(file: File | string): string | null { } export class ElectronImportAdapter implements ImportAdapter { + private activeBatchId: string | null = null + async importFile( file: File | string, options?: ImportOptions, @@ -75,6 +80,77 @@ export class ElectronImportAdapter implements ImportAdapter { }) } + async importBatch( + items: BatchImportItem[], + options?: ImportOptions, + onProgress?: (progress: BatchImportProgress) => void + ): Promise { + const resolvedItems: Array<{ id: string; filePath: string }> = [] + for (const item of items) { + const filePath = resolveFilePath(item.file) + if (!filePath) { + return items.map((candidate) => + candidate.id === item.id + ? { id: candidate.id, status: 'failed' as const, error: 'Cannot get file path in Electron' } + : { id: candidate.id, status: 'cancelled' as const } + ) + } + resolvedItems.push({ id: item.id, filePath }) + } + + const batchId = crypto.randomUUID() + this.activeBatchId = batchId + const unlisten = window.chatApi.onImportBatchProgress((event) => { + if (event.batchId !== batchId) return + const index = event.batchIndex + onProgress?.({ + index, + event: event.batchEvent, + progress: + event.batchEvent === 'progress' + ? { + stage: event.stage as ImportProgress['stage'], + progress: event.percentage, + message: event.message ?? '', + } + : undefined, + result: event.batchResult as BatchImportItemResult | undefined, + }) + }) + + try { + const results = await window.chatApi.importBatch(batchId, resolvedItems, { + sessionGapThreshold: options?.sessionGapThreshold, + }) + return results.map((item) => { + if (item.status === 'cancelled') return { id: item.id, status: 'cancelled' } + if (item.status === 'failed') { + return { + id: item.id, + status: 'failed', + error: item.error ?? item.result?.error ?? 'error.import_failed', + result: item.result ? normalizeImportResult(item.result) : undefined, + } + } + return { + id: item.id, + status: 'success', + result: normalizeImportResult(item.result ?? { success: false, error: 'error.import_failed' }), + } + }) + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + return items.map((item) => ({ id: item.id, status: 'failed', error: message })) + } finally { + unlisten() + if (this.activeBatchId === batchId) this.activeBatchId = null + } + } + + cancelActiveImport(): void { + if (this.activeBatchId) void window.chatApi.cancelImportBatch(this.activeBatchId) + } + async detectFormat(file: File | string): Promise { const filePath = resolveFilePath(file) if (!filePath) return null diff --git a/src/services/import/fetch.ts b/src/services/import/fetch.ts index 4bf60003..015129c7 100644 --- a/src/services/import/fetch.ts +++ b/src/services/import/fetch.ts @@ -16,6 +16,9 @@ import type { DemoImportResult, IncrementalAnalysis, IncrementalImportResult, + BatchImportItem, + BatchImportItemResult, + BatchImportProgress, } from './types' import { normalizeImportResult } from './types' import { get, fetchWithAuth, getBaseUrl } from '../utils/http' @@ -56,6 +59,18 @@ async function consumeSseStream(res: Response, fallback: T, onProgress?: (p: } export class FetchImportAdapter implements ImportAdapter { + private activeBatch: { id: string; cancelRequested: boolean } | null = null + + private async requestBatchCancellation(batchId: string): Promise { + try { + await fetchWithAuth(`${getBaseUrl()}/import/batch/${encodeURIComponent(batchId)}/cancel`, { + method: 'POST', + }) + } catch { + // The import stream remains authoritative and will report the final item states. + } + } + async importFile( file: File | string, options?: ImportOptions, @@ -85,6 +100,108 @@ export class FetchImportAdapter implements ImportAdapter { ) } + async importBatch( + items: BatchImportItem[], + options?: ImportOptions, + onProgress?: (progress: BatchImportProgress) => void + ): Promise { + if (items.some((item) => typeof item.file === 'string')) { + return items.map((item) => ({ + id: item.id, + status: 'failed', + error: 'File path import is not supported in Web mode', + })) + } + + const form = new FormData() + for (const item of items) form.append('files', item.file as File) + if (options?.sessionGapThreshold !== undefined) { + form.append('sessionGapThreshold', String(options.sessionGapThreshold)) + } + + const activeBatch = { id: crypto.randomUUID(), cancelRequested: false } + this.activeBatch = activeBatch + const completedResults: Array = Array(items.length) + try { + const response = await fetchWithAuth(`${getBaseUrl()}/import/batch`, { + method: 'POST', + body: form, + headers: { 'X-ChatLab-Import-Batch-Id': activeBatch.id }, + }) + if (activeBatch.cancelRequested) await this.requestBatchCancellation(activeBatch.id) + if (!response.ok || !response.body) { + const text = await response.text() + return items.map((item) => ({ + id: item.id, + status: 'failed', + error: `HTTP ${response.status}: ${text}`, + })) + } + + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + let results: BatchImportItemResult[] | null = null + let eventType = '' + + while (true) { + const { done, value } = await reader.read() + if (done) break + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() || '' + for (const line of lines) { + if (line.startsWith('event: ')) { + eventType = line.slice(7).trim() + continue + } + if (!line.startsWith('data: ')) continue + const data = JSON.parse(line.slice(6)) + if (eventType === 'batch-start') { + onProgress?.({ index: data.index, event: 'start' }) + } else if (eventType === 'batch-progress') { + onProgress?.({ index: data.index, event: 'progress', progress: data.progress }) + } else if (eventType === 'batch-complete') { + const result = normalizeBatchResult(items, data.index, data.result) + completedResults[data.index] = result + onProgress?.({ index: data.index, event: 'complete', result }) + } else if (eventType === 'done') { + results = (data as unknown[]).map((result, index) => normalizeBatchResult(items, index, result)) + } + eventType = '' + } + } + return ( + results ?? + items.map( + (item, index): BatchImportItemResult => + completedResults[index] ?? { + id: item.id, + status: 'failed', + error: 'Batch response ended without results', + } + ) + ) + } catch (error) { + return items.map( + (item, index): BatchImportItemResult => + completedResults[index] ?? { + id: item.id, + status: 'failed', + error: error instanceof Error ? error.message : String(error), + } + ) + } finally { + if (this.activeBatch === activeBatch) this.activeBatch = null + } + } + + cancelActiveImport(): void { + if (!this.activeBatch) return + this.activeBatch.cancelRequested = true + void this.requestBatchCancellation(this.activeBatch.id) + } + async detectFormat(file: File | string): Promise { if (typeof file === 'string') return null const form = new FormData() @@ -291,3 +408,17 @@ export class FetchImportAdapter implements ImportAdapter { ) } } + +function normalizeBatchResult(items: BatchImportItem[], index: number, raw: any): BatchImportItemResult { + const id = items[index]?.id ?? String(index) + if (raw?.status === 'cancelled') return { id, status: 'cancelled' } + if (raw?.status === 'success') { + return { id, status: 'success', result: normalizeImportResult(raw.result) } + } + return { + id, + status: 'failed', + error: raw?.error ?? raw?.result?.error ?? 'error.import_failed', + result: raw?.result ? normalizeImportResult(raw.result) : undefined, + } +} diff --git a/src/services/import/importSourceAdapters.test.ts b/src/services/import/importSourceAdapters.test.ts index 6bf360e7..20805fd4 100644 --- a/src/services/import/importSourceAdapters.test.ts +++ b/src/services/import/importSourceAdapters.test.ts @@ -17,6 +17,88 @@ afterEach(() => { }) describe('archive import source adapters', () => { + it('uses the Electron backend batch API and forwards cancellation', async () => { + const calls: unknown[][] = [] + let progressCallback: ((progress: any) => void) | undefined + Object.defineProperty(globalThis, 'window', { + configurable: true, + value: { + chatApi: { + importBatch: async (...args: unknown[]) => { + calls.push(['batch', ...args]) + progressCallback?.({ + batchId: args[0], + batchIndex: 0, + batchEvent: 'complete', + stage: 'done', + percentage: 100, + batchResult: { id: '0', status: 'success', result: { success: true, sessionId: 'session-1' } }, + }) + return [{ id: '0', status: 'success', result: { success: true, sessionId: 'session-1' } }] + }, + cancelImportBatch: async (...args: unknown[]) => { + calls.push(['cancel', ...args]) + return { success: true } + }, + onImportBatchProgress: (callback: (progress: any) => void) => { + progressCallback = callback + return () => {} + }, + }, + }, + }) + + const adapter = new ElectronImportAdapter() + const progress: unknown[] = [] + const promise = adapter.importBatch?.( + [{ id: '0', file: '/tmp/first.json' }], + { sessionGapThreshold: 7200 }, + (event) => progress.push(event) + ) + adapter.cancelActiveImport?.() + const result = await promise + + assert.equal(result?.[0].status, 'success') + assert.equal((result?.[0] as any).result.sessionId, 'session-1') + assert.equal(calls[0][0], 'batch') + assert.deepEqual( + (calls[0][2] as Array<{ filePath: string }>).map((item) => item.filePath), + ['/tmp/first.json'] + ) + assert.deepEqual(calls[0][3], { sessionGapThreshold: 7200 }) + assert.equal(calls[1][0], 'cancel') + assert.equal(progress.length, 1) + }) + + it('maps an Electron batch IPC failure to per-item failures', async () => { + let unlistenCount = 0 + Object.defineProperty(globalThis, 'window', { + configurable: true, + value: { + chatApi: { + importBatch: async () => { + throw new Error('worker unavailable') + }, + onImportBatchProgress: () => () => { + unlistenCount++ + }, + }, + }, + }) + + const adapter = new ElectronImportAdapter() + const results = await adapter.importBatch?.([ + { id: 'first', file: '/tmp/first.json' }, + { id: 'second', file: '/tmp/second.json' }, + ]) + + assert.deepEqual(results, [ + { id: 'first', status: 'failed', error: 'worker unavailable' }, + { id: 'second', status: 'failed', error: 'worker unavailable' }, + ]) + assert.equal(unlistenCount, 1) + }) + it('forwards the session gap threshold through Electron file import', async () => { const calls: unknown[][] = [] Object.defineProperty(globalThis, 'window', { @@ -142,6 +224,110 @@ describe('archive import source adapters', () => { assert.equal(requests[2].init?.method, 'DELETE') }) + it('uploads Web batch files once and maps SSE item results', async () => { + let request: { url: string; init?: RequestInit } | undefined + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + request = { url: String(input), init } + return new Response( + [ + 'event: batch-start', + 'data: {"index":0}', + '', + 'event: batch-complete', + 'data: {"index":0,"result":{"id":"0","status":"success","result":{"success":true,"sessionId":"session-a"}}}', + '', + 'event: done', + 'data: [{"id":"0","status":"success","result":{"success":true,"sessionId":"session-a"}},{"id":"1","status":"failed","error":"bad file"}]', + '', + '', + ].join('\n'), + { headers: { 'Content-Type': 'text/event-stream' } } + ) + }) as typeof fetch + + const progress: unknown[] = [] + const result = await new FetchImportAdapter().importBatch?.( + [ + { id: 'first', file: new File(['{}'], 'first.json') }, + { id: 'second', file: new File(['{}'], 'second.json') }, + ], + { sessionGapThreshold: 7200 }, + (event) => progress.push(event) + ) + + assert.equal(request?.url, '/_web/import/batch') + assert.equal(request?.init?.body instanceof FormData, true) + const form = request?.init?.body as FormData + assert.equal(form.getAll('files').length, 2) + assert.equal(form.get('sessionGapThreshold'), '7200') + assert.deepEqual( + result?.map((item) => ({ id: item.id, status: item.status })), + [ + { id: 'first', status: 'success' }, + { id: 'second', status: 'failed' }, + ] + ) + assert.equal(progress.length, 2) + }) + + it('keeps the Web batch stream open until cancellation returns authoritative results', async () => { + const requests: Array<{ url: string; batchId?: string }> = [] + let streamController: ReadableStreamDefaultController | undefined + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input) + if (url.includes('/cancel')) { + requests.push({ url }) + streamController?.enqueue( + new TextEncoder().encode( + [ + 'event: batch-complete', + 'data: {"index":0,"result":{"id":"0","status":"success","result":{"success":true,"sessionId":"session-a"}}}', + '', + 'event: batch-complete', + 'data: {"index":1,"result":{"id":"1","status":"cancelled"}}', + '', + 'event: done', + 'data: [{"id":"0","status":"success","result":{"success":true,"sessionId":"session-a"}},{"id":"1","status":"cancelled"}]', + '', + '', + ].join('\n') + ) + ) + streamController?.close() + return new Response(JSON.stringify({ success: true, active: true })) + } + + const batchId = new Headers(init?.headers).get('X-ChatLab-Import-Batch-Id') ?? undefined + requests.push({ url, batchId }) + const stream = new ReadableStream({ + start(controller) { + streamController = controller + }, + }) + return new Response(stream, { headers: { 'Content-Type': 'text/event-stream' } }) + }) as typeof fetch + + const adapter = new FetchImportAdapter() + const resultPromise = adapter.importBatch?.([ + { id: 'first', file: new File(['{}'], 'first.json') }, + { id: 'second', file: new File(['{}'], 'second.json') }, + ]) + await new Promise((resolve) => setTimeout(resolve, 0)) + adapter.cancelActiveImport?.() + const results = await resultPromise + + assert.deepEqual( + results?.map((result) => ({ id: result.id, status: result.status })), + [ + { id: 'first', status: 'success' }, + { id: 'second', status: 'cancelled' }, + ] + ) + assert.equal(requests.length, 2) + assert.ok(requests[0].batchId) + assert.equal(requests[1].url, `/_web/import/batch/${requests[0].batchId}/cancel`) + }) + it('sends the browser timezone when importing Demo sessions through CLI Web', async () => { let requestBody: string | undefined globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { diff --git a/src/services/import/types.ts b/src/services/import/types.ts index 2aa08c45..c1a9092a 100644 --- a/src/services/import/types.ts +++ b/src/services/import/types.ts @@ -165,6 +165,23 @@ export interface IncrementalImportResult { error?: string } +export interface BatchImportItem { + id: string + file: File | string +} + +export type BatchImportItemResult = + | { id: string; status: 'success'; result: ImportResult } + | { id: string; status: 'failed'; error: string; result?: ImportResult } + | { id: string; status: 'cancelled' } + +export interface BatchImportProgress { + index: number + event: 'start' | 'progress' | 'complete' + progress?: ImportProgress + result?: BatchImportItemResult +} + // ==================== 核心接口 ==================== export interface ImportAdapter { @@ -177,6 +194,16 @@ export interface ImportAdapter { onProgress?: (p: ImportProgress) => void ): Promise + /** + * Import multiple sources under one backend-owned coordinator/lock lease. + * Adapters without a safe batch runtime may omit this capability. + */ + importBatch?( + items: BatchImportItem[], + options?: ImportOptions, + onProgress?: (progress: BatchImportProgress) => void + ): Promise + /** Cancel the active import when the runtime supports cooperative cancellation. */ cancelActiveImport?(): void diff --git a/src/services/index.ts b/src/services/index.ts index b88bd042..3b7b195a 100644 --- a/src/services/index.ts +++ b/src/services/index.ts @@ -39,6 +39,9 @@ export type { DemoImportResult, IncrementalAnalysis, IncrementalImportResult, + BatchImportItem, + BatchImportItemResult, + BatchImportProgress, } from './import/types' export { useSessionIndexService } from './session-index/service' export { useSemanticIndexService } from './semantic-index/service' diff --git a/src/stores/session-batch-import.test.ts b/src/stores/session-batch-import.test.ts index bee2694a..b2ce3be5 100644 --- a/src/stores/session-batch-import.test.ts +++ b/src/stores/session-batch-import.test.ts @@ -45,7 +45,7 @@ test('batch import accepts browser File objects and processes them serially', as duplicateCount: 0, } }, - } as ImportAdapter) + } as unknown as ImportAdapter) registerAdapter('data', { getSessions: async () => [], tryApplyOwnerProfile: async () => ({ applied: false }), @@ -76,3 +76,55 @@ test('batch import accepts browser File objects and processes them serially', as ] ) }) + +test('batch import delegates scheduling and cancellation to a capable backend adapter', async () => { + Object.defineProperties(globalThis, { + localStorage: { configurable: true, value: createMemoryStorage() }, + sessionStorage: { configurable: true, value: createMemoryStorage() }, + }) + + let cancelCalls = 0 + let receivedFiles: Array = [] + registerAdapter('import', { + async importFile() { + throw new Error('single-file fallback should not run') + }, + async importBatch(items, _options, onProgress) { + receivedFiles = items.map((item) => item.file) + onProgress?.({ index: 0, event: 'start' }) + const first = { + id: items[0].id, + status: 'success' as const, + result: { success: true, sessionId: 'session-1', importMode: 'created' as const }, + } + onProgress?.({ index: 0, event: 'complete', result: first }) + return [first, { id: items[1].id, status: 'cancelled' as const }] + }, + cancelActiveImport() { + cancelCalls++ + }, + } as unknown as ImportAdapter) + registerAdapter('data', { + getSessions: async () => [], + tryApplyOwnerProfile: async () => ({ applied: false }), + } as unknown as DataAdapter) + + setActivePinia(createPinia()) + const { useSessionStore } = await import('./session') + const store = useSessionStore() + const files = [new File(['a'], 'first.json'), new File(['b'], 'second.json')] + const result = await store.importFiles(files) + store.cancelBatchImport() + + assert.deepEqual( + receivedFiles.map((file) => (file as File).name), + ['first.json', 'second.json'] + ) + assert.equal(result.success, 1) + assert.equal(result.cancelled, 1) + assert.deepEqual( + result.files.map((file) => file.status), + ['success', 'cancelled'] + ) + assert.equal(cancelCalls, 1) +}) diff --git a/src/stores/session.ts b/src/stores/session.ts index e9576e23..db2a13f8 100644 --- a/src/stores/session.ts +++ b/src/stores/session.ts @@ -357,6 +357,66 @@ export const useSessionStore = defineStore( status: 'pending' as BatchFileStatus, })) + const importService = useImportService() + if (importService.importBatch) { + const results = await importService.importBatch( + importSources.map(({ source }, index) => ({ id: String(index), file: source })), + { sessionGapThreshold: getSessionGapThreshold() }, + ({ index, event, progress, result }) => { + const file = batchFiles.value[index] + if (!file) return + if (event === 'start') file.status = 'importing' + if (progress) file.progress = progress + if (event === 'complete' && result) { + file.status = result.status + if (result.status === 'success') { + file.sessionId = result.result.sessionId + file.importMode = result.result.importMode + file.matchedBy = result.result.matchedBy + file.createReason = result.result.createReason + file.newMessageCount = result.result.newMessageCount + file.duplicateCount = result.result.duplicateCount + } else if (result.status === 'failed') { + file.error = result.error + } + } + } + ) + + for (let index = 0; index < results.length; index++) { + const result = results[index] + const file = batchFiles.value[index] + file.status = result.status + if (result.status === 'success') { + file.sessionId = result.result.sessionId + file.importMode = result.result.importMode + file.matchedBy = result.result.matchedBy + file.createReason = result.result.createReason + file.newMessageCount = result.result.newMessageCount + file.duplicateCount = result.result.duplicateCount + } else if (result.status === 'failed') { + file.error = result.error + } + } + + const successfulSessionIds = results.flatMap((result) => + result.status === 'success' && result.result.sessionId ? [result.result.sessionId] : [] + ) + for (const sessionId of successfulSessionIds) await applyOwnerProfileAfterImport(sessionId) + await loadSessions() + + const result: BatchImportResult = { + total: sources.length, + success: results.filter((item) => item.status === 'success').length, + failed: results.filter((item) => item.status === 'failed').length, + cancelled: results.filter((item) => item.status === 'cancelled').length, + files: [...batchFiles.value], + } + batchImportResult.value = result + isBatchImporting.value = false + return result + } + let successCount = 0 let failedCount = 0 let cancelledCount = 0 @@ -509,6 +569,7 @@ export const useSessionStore = defineStore( */ function cancelBatchImport() { batchImportCancelled.value = true + useImportService().cancelActiveImport?.() } /**