From 61d42b25009d98a72ea680141c1062bfa7ec99f0 Mon Sep 17 00:00:00 2001 From: digua Date: Sun, 12 Jul 2026 23:15:37 +0800 Subject: [PATCH] fix(import): harden automatic incremental imports --- apps/cli/src/cli.ts | 13 +- apps/cli/src/http/routes/web/import.test.ts | 65 +++++++ apps/cli/src/http/routes/web/index.ts | 5 + apps/cli/src/import/importer.test.ts | 65 ++++++- apps/cli/src/import/stream-import.ts | 96 ++++++++--- apps/desktop/main/api/errors.test.ts | 8 + apps/desktop/main/api/errors.ts | 2 + apps/desktop/main/internal-api.ts | 44 ++--- apps/desktop/main/worker/workerManager.ts | 88 +++++++--- apps/desktop/preload/apis/chat.ts | 2 + apps/desktop/preload/index.d.ts | 3 + docs/cn/standard/chatlab-import.md | 2 +- docs/en/standard/chatlab-import.md | 2 +- packages/http-routes/src/errors.test.ts | 8 + packages/http-routes/src/errors.ts | 2 + packages/node-runtime/src/database-manager.ts | 4 + .../src/import/auto-import-matcher.test.ts | 159 ++++++++++++++++++ .../src/import/auto-import-matcher.ts | 26 ++- .../src/import/auto-importer.test.ts | 2 + .../node-runtime/src/import/auto-importer.ts | 9 +- .../src/import/import-lock.test.ts | 82 +++++++++ .../node-runtime/src/import/import-lock.ts | 131 +++++++++++++++ .../src/import/incremental-importer.test.ts | 45 +++++ .../src/import/incremental-importer.ts | 45 +++-- packages/node-runtime/src/import/index.ts | 13 +- .../src/import/message-deduplicator.ts | 47 ++++++ .../src/import/streaming-importer.test.ts | 80 +++++++++ .../src/import/streaming-importer.ts | 33 +++- packages/node-runtime/src/index.ts | 5 + .../src/services/push-importer.test.ts | 22 +++ .../src/services/push-importer.ts | 29 ++-- src/i18n/locales/en-US/home.json | 3 + src/i18n/locales/ja-JP/home.json | 3 + src/i18n/locales/zh-CN/home.json | 3 + src/i18n/locales/zh-TW/home.json | 3 + .../home/components/import/ImportArea.vue | 21 ++- .../import/preparedImportFlow.test.ts | 3 + .../components/import/preparedImportFlow.ts | 4 +- src/services/import/electron.ts | 2 + src/services/import/types.ts | 4 + src/services/index.ts | 1 + src/stores/session.ts | 9 +- 42 files changed, 1075 insertions(+), 118 deletions(-) create mode 100644 packages/node-runtime/src/import/import-lock.test.ts create mode 100644 packages/node-runtime/src/import/import-lock.ts create mode 100644 packages/node-runtime/src/import/message-deduplicator.ts diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts index 4a91c106..ce4b0518 100644 --- a/apps/cli/src/cli.ts +++ b/apps/cli/src/cli.ts @@ -13,6 +13,7 @@ import { initAppLogger, appLogger, getSystemLogsDir, + IMPORT_IN_PROGRESS_ERROR_KEY, logNativeParserStatus, } from '@openchatlab/node-runtime' import { getVersion } from './version' @@ -95,8 +96,12 @@ program if (result.success) { console.log(`\n\nImport succeeded!`) console.log(` Session ID: ${result.sessionId}`) - console.log(` Mode: ${result.importMode === 'incremental' ? 'incremental' : 'created'}`) + console.log(` Mode: ${result.importMode ?? 'unknown'}`) if (result.matchedBy) console.log(` Matched by: ${result.matchedBy}`) + if (result.createReason) console.log(` Create reason: ${result.createReason}`) + if (result.createReason === 'ambiguous') { + console.warn(' Warning: multiple existing sessions matched, so a new session was created.') + } console.log(` New messages: ${result.newMessageCount ?? 0}`) console.log(` Duplicates skipped: ${result.duplicateCount ?? 0}`) @@ -117,7 +122,11 @@ program } } } else { - console.error(`\n\nImport failed: ${result.error}`) + const errorMessage = + result.error === IMPORT_IN_PROGRESS_ERROR_KEY + ? 'Another import is already in progress. Please retry later.' + : result.error + console.error(`\n\nImport failed: ${errorMessage}`) process.exitCode = 1 } } catch (err) { diff --git a/apps/cli/src/http/routes/web/import.test.ts b/apps/cli/src/http/routes/web/import.test.ts index 469c9abd..863527cf 100644 --- a/apps/cli/src/http/routes/web/import.test.ts +++ b/apps/cli/src/http/routes/web/import.test.ts @@ -71,4 +71,69 @@ describe('CLI Web automatic import route', () => { await app.close() } }) + + it('returns a stable import-in-progress error event instead of an internal database error', async () => { + const app = Fastify() + try { + await app.register(multipart) + registerImportRoutes( + app, + {} as any, + { + runAutoImport: async () => ({ success: false, error: 'error.import_in_progress' }), + } as any + ) + + const body = multipartPayload() + const response = await app.inject({ + method: 'POST', + url: '/_web/import', + headers: { 'content-type': body.contentType }, + payload: body.payload, + }) + + assert.equal(response.statusCode, 200) + assert.match(response.body, /event: error/) + assert.match(response.body, /"error":"error\.import_in_progress"/) + assert.doesNotMatch(response.body, /no such column|SQLITE/i) + } finally { + await app.close() + } + }) + + it('preserves an ambiguous create reason in the done event', async () => { + const app = Fastify() + try { + await app.register(multipart) + registerImportRoutes( + app, + {} as any, + { + runAutoImport: async () => ({ + success: true, + sessionId: 'new-session', + importMode: 'created', + createReason: 'ambiguous', + newMessageCount: 3, + duplicateCount: 0, + }), + } as any + ) + + const body = multipartPayload() + const response = await app.inject({ + method: 'POST', + url: '/_web/import', + headers: { 'content-type': body.contentType }, + payload: body.payload, + }) + + assert.equal(response.statusCode, 200) + assert.match(response.body, /event: done/) + assert.match(response.body, /"importMode":"created"/) + assert.match(response.body, /"createReason":"ambiguous"/) + } finally { + await app.close() + } + }) }) diff --git a/apps/cli/src/http/routes/web/index.ts b/apps/cli/src/http/routes/web/index.ts index 9065306d..d28b41d4 100644 --- a/apps/cli/src/http/routes/web/index.ts +++ b/apps/cli/src/http/routes/web/index.ts @@ -98,6 +98,11 @@ export function registerWebRoutes( const cliStreamImport = async (dm: typeof dbManager, filePath: string) => { const { streamImport } = await import('../../../import/stream-import') const result = await streamImport(dm, filePath) + if (!result.success) { + const { IMPORT_IN_PROGRESS_ERROR_KEY, ImportInProgressError } = await import('@openchatlab/node-runtime') + if (result.error === IMPORT_IN_PROGRESS_ERROR_KEY) throw new ImportInProgressError() + throw new Error(result.error || 'Import failed') + } if (!result.sessionId) throw new Error('Import succeeded but no sessionId returned') return { sessionId: result.sessionId } } diff --git a/apps/cli/src/import/importer.test.ts b/apps/cli/src/import/importer.test.ts index 6afd844d..1b07ef00 100644 --- a/apps/cli/src/import/importer.test.ts +++ b/apps/cli/src/import/importer.test.ts @@ -4,7 +4,13 @@ import os from 'node:os' import path from 'node:path' import test from 'node:test' import type { PathProvider } from '@openchatlab/core' -import { DatabaseManager, raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta } from '@openchatlab/node-runtime' +import { + DatabaseManager, + IMPORT_IN_PROGRESS_ERROR_KEY, + raiseDataDirMinRuntimeVersion, + readDataDirCompatibilityMeta, + withDataDirImportLock, +} from '@openchatlab/node-runtime' import { autoImport, streamImport } from './stream-import' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -45,6 +51,21 @@ function writeTempChatFile(dir: string): string { return filePath } +function writeTempDuplicateChatFile(dir: string): string { + const filePath = path.join(dir, 'duplicate-chat.json') + const message = { sender: 'u1', accountName: 'Alice', timestamp: 1711468800, type: 0, content: 'hello' } + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 1711468800 }, + meta: { name: 'Duplicate Chat', platform: 'qq', type: 'group', groupId: 'duplicate-group' }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + messages: [message, { ...message }], + }) + ) + 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 }) @@ -115,6 +136,32 @@ test('autoImport creates once and then incrementally imports the same stable cha assert.equal(manager.listSessionIds().length, 1) }) +test('autoImport reports the same exact-message deduplication on create and incremental paths', 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 chatFile = writeTempDuplicateChatFile(root) + + const created = await autoImport(manager, chatFile, { nativeBinding }) + const incremental = await autoImport(manager, chatFile, { nativeBinding }) + + assert.equal(created.success, true) + assert.equal(created.newMessageCount, 1) + assert.equal(created.duplicateCount, 1) + assert.equal(incremental.success, true) + assert.equal(incremental.newMessageCount, 0) + assert.equal(incremental.duplicateCount, 2) + + const db = manager.openRawSessionDatabase(created.sessionId!, { readonly: true }) + const row = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number } + db.close() + assert.equal(row.count, 1) +}) + test('autoImport uses an explicit id for create and then forces incremental import', async () => { const root = makeTempDir() fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) @@ -136,3 +183,19 @@ test('autoImport uses an explicit id for create and then forces incremental impo assert.equal(incremental.newMessageCount, 0) assert.equal(manager.listSessionIds().length, 1) }) + +test('autoImport rejects a concurrent writer before opening a session database', 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 chatFile = writeTempChatFile(root) + + const result = await withDataDirImportLock(manager.getUserDataDir(), () => autoImport(manager, chatFile)) + + assert.deepEqual(result, { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY }) + assert.equal(fs.readdirSync(path.join(root, 'data', 'databases')).length, 0) +}) diff --git a/apps/cli/src/import/stream-import.ts b/apps/cli/src/import/stream-import.ts index 34213673..5d813167 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -9,11 +9,14 @@ import type { DatabaseManager } from '@openchatlab/node-runtime' import { DataDirCompatibilityError, + IMPORT_IN_PROGRESS_ERROR_KEY, + ImportInProgressError, autoImportFile as sharedAutoImportFile, streamingImport, analyzeNewImport as sharedAnalyzeNewImport, analyzeIncrementalImport as sharedAnalyzeIncremental, incrementalImport as sharedIncrementalImport, + withDataDirImportLock, } from '@openchatlab/node-runtime' import type { StreamImportResult, @@ -128,7 +131,7 @@ function deleteSessionDatabase(dbManager: DatabaseManager, sessionId: string): v * High-performance streaming import: parse a file and write to DB * with batched transactions, deferred indexes, and FTS. */ -export async function streamImport( +async function streamImportUnlocked( dbManager: DatabaseManager, filePath: string, options?: StreamImportOptions @@ -159,6 +162,23 @@ export async function streamImport( return result } +export async function streamImport( + dbManager: DatabaseManager, + filePath: string, + options?: StreamImportOptions +): Promise { + try { + return await withDataDirImportLock(dbManager.getUserDataDir(), () => + streamImportUnlocked(dbManager, filePath, options) + ) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY } + } + throw error + } +} + export async function autoImport( dbManager: DatabaseManager, filePath: string, @@ -170,30 +190,40 @@ 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) => - streamImport(dbManager, sourcePath, { - ...options, - ...sourceFormatOptions, - sessionId: explicitSessionId, - }), - appendSession: (targetSessionId, sourcePath, sourceFormatOptions) => - incrementalImport(dbManager, targetSessionId, sourcePath, { - ...sourceFormatOptions, + try { + return await withDataDirImportLock(dbManager.getUserDataDir(), () => + sharedAutoImportFile( + filePath, + { + listSessionIds: () => dbManager.listSessionIds(), + openReadonly: (candidateSessionId) => + dbManager.openRawSessionDatabase(candidateSessionId, { readonly: true }), onProgress: progressAdapter, - }), - }, - { - explicitSessionId: sessionId, - formatOptions, + sessionExists: (candidateSessionId) => dbManager.listSessionIds().includes(candidateSessionId), + createSession: (sourcePath, sourceFormatOptions, explicitSessionId) => + streamImportUnlocked(dbManager, sourcePath, { + ...options, + ...sourceFormatOptions, + sessionId: explicitSessionId, + }), + appendSession: (targetSessionId, sourcePath, sourceFormatOptions) => + incrementalImportUnlocked(dbManager, targetSessionId, sourcePath, { + ...sourceFormatOptions, + onProgress: progressAdapter, + }), + }, + { + explicitSessionId: sessionId, + formatOptions, + } + ) + ) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY } } - ) + throw error + } } // ==================== Incremental import ==================== @@ -216,7 +246,7 @@ function buildIncrementalDeps( } } -export async function incrementalImport( +async function incrementalImportUnlocked( dbManager: DatabaseManager, sessionId: string, filePath: string, @@ -249,6 +279,24 @@ export async function incrementalImport( return result } +export async function incrementalImport( + dbManager: DatabaseManager, + sessionId: string, + filePath: string, + options?: ImportOptions & { onProgress?: ImportProgressCallback } +): Promise { + try { + return await withDataDirImportLock(dbManager.getUserDataDir(), () => + incrementalImportUnlocked(dbManager, sessionId, filePath, options) + ) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { success: false, newMessageCount: 0, error: IMPORT_IN_PROGRESS_ERROR_KEY } + } + throw error + } +} + export async function analyzeIncrementalImport( dbManager: DatabaseManager, sessionId: string, diff --git a/apps/desktop/main/api/errors.test.ts b/apps/desktop/main/api/errors.test.ts index 3b33e5f9..e64e9fa4 100644 --- a/apps/desktop/main/api/errors.test.ts +++ b/apps/desktop/main/api/errors.test.ts @@ -2,6 +2,14 @@ import assert from 'node:assert/strict' import test from 'node:test' import { DataDirCompatibilityError } from '@openchatlab/node-runtime/src/data-dir-compat' import { ApiErrorCode, apiErrorFromUnknown } from './errors' +import { ImportInProgressError } from '@openchatlab/node-runtime/src/import/import-lock' + +test('apiErrorFromUnknown maps the shared import lock error to 409', () => { + const apiError = apiErrorFromUnknown(new ImportInProgressError()) + + assert.equal(apiError?.code, ApiErrorCode.IMPORT_IN_PROGRESS) + assert.equal(apiError?.statusCode, 409) +}) test('apiErrorFromUnknown maps wrapped data directory compatibility errors to 409', () => { const cause = new DataDirCompatibilityError( diff --git a/apps/desktop/main/api/errors.ts b/apps/desktop/main/api/errors.ts index 34f6b530..a2bbb59a 100644 --- a/apps/desktop/main/api/errors.ts +++ b/apps/desktop/main/api/errors.ts @@ -3,6 +3,7 @@ */ import { DataDirCompatibilityError } from '@openchatlab/node-runtime/src/data-dir-compat' +import { ImportInProgressError } from '@openchatlab/node-runtime/src/import/import-lock' export enum ApiErrorCode { UNAUTHORIZED = 'UNAUTHORIZED', @@ -103,6 +104,7 @@ export function serverError(message = 'Internal server error'): ApiError { export function apiErrorFromUnknown(error: unknown): ApiError | null { if (error instanceof ApiError) return error + if (error instanceof ImportInProgressError) return importInProgress() const compatibilityError = findDataDirCompatibilityError(error) if (compatibilityError) return dataDirIncompatible(compatibilityError.message) return null diff --git a/apps/desktop/main/internal-api.ts b/apps/desktop/main/internal-api.ts index 5b378412..91eaad6c 100644 --- a/apps/desktop/main/internal-api.ts +++ b/apps/desktop/main/internal-api.ts @@ -21,6 +21,7 @@ import { CustomProviderStore, CustomModelStore, MergeSessionCache, + withDataDirImportLock, raiseChatDbCompatibilityGate, streamingImport, createSemanticIndexWorkerRuntimeClient, @@ -138,26 +139,29 @@ export async function startInternalServer(pathProvider: PathProvider): Promise { - const deps: StreamImportDeps = { - openDatabase(sessionId: string) { - return dm.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true }) - }, - deleteDatabase(sessionId: string) { - dm.deleteSessionDatabaseFiles(sessionId) - }, - onProgress() { - /* no progress for merge-triggered import */ - }, - } - const result = await streamingImport(filePath, deps) - if (!result.sessionId) throw new Error('Import succeeded but no sessionId returned') - try { - raiseChatDbCompatibilityGate(pathProvider, runtime) - } catch (error) { - dm.deleteSessionDatabaseFiles(result.sessionId) - throw error - } - return { sessionId: result.sessionId } + return withDataDirImportLock(dm.getUserDataDir(), async () => { + const deps: StreamImportDeps = { + openDatabase(sessionId: string) { + return dm.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true }) + }, + deleteDatabase(sessionId: string) { + dm.deleteSessionDatabaseFiles(sessionId) + }, + onProgress() { + /* no progress for merge-triggered import */ + }, + } + const result = await streamingImport(filePath, deps) + if (!result.success) throw new Error(result.error || 'Import failed') + if (!result.sessionId) throw new Error('Import succeeded but no sessionId returned') + try { + raiseChatDbCompatibilityGate(pathProvider, runtime) + } catch (error) { + dm.deleteSessionDatabaseFiles(result.sessionId) + throw error + } + return { sessionId: result.sessionId } + }) } const { shell } = await import('electron') diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index 760af638..b03bba07 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/apps/desktop/main/worker/workerManager.ts @@ -15,7 +15,12 @@ import { getNlpDir } from '../nlp/dictManager' import { resolveDesktopNativeBinding } from '../native-sqlite' import { assertDesktopDataDirCompatible, getDesktopAppVersion } from '../runtime-compat' import { getPathProvider } from '../path-context' -import { raiseChatDbCompatibilityGate } from '@openchatlab/node-runtime' +import { + IMPORT_IN_PROGRESS_ERROR_KEY, + ImportInProgressError, + raiseChatDbCompatibilityGate, + withDataDirImportLock, +} from '@openchatlab/node-runtime' import { isRestartableReadOnlyRequestType } from './workerTimeoutPolicy' interface WorkerRequestOptions { @@ -519,7 +524,7 @@ export async function streamParseFileInfo( * @param onProgress 进度回调 * @param formatOptions 格式特定选项（如 Telegram 的 chatIndex） */ -export async function streamImport( +async function streamImportUnlocked( filePath: string, onProgress?: (progress: ParseProgress) => void, formatOptions?: Record, @@ -551,36 +556,63 @@ export async function streamImport( return result } +export async function streamImport( + filePath: string, + onProgress?: (progress: ParseProgress) => void, + formatOptions?: Record, + externalSessionId?: string +): Promise { + try { + return await withDataDirImportLock(getPathProvider().getUserDataDir(), () => + streamImportUnlocked(filePath, onProgress, formatOptions, externalSessionId) + ) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY } + } + throw error + } +} + export async function autoImport( filePath: string, onProgress?: (progress: ParseProgress) => void, formatOptions?: Record, explicitSessionId?: string ): Promise { - assertDataDirCompatibleNow() - - const result = await sendToWorkerWithProgress( - 'autoImport', - { filePath, formatOptions, explicitSessionId }, - onProgress - ) - if (!result.success || !result.sessionId || result.importMode !== 'created') return result - try { - raiseChatDbCompatibilityGate(getPathProvider(), { - version: getDesktopAppVersion(app.getVersion()), - kind: 'desktop', + return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => { + assertDataDirCompatibleNow() + + const result = await sendToWorkerWithProgress( + 'autoImport', + { filePath, formatOptions, explicitSessionId }, + onProgress + ) + if (!result.success || !result.sessionId || result.importMode !== 'created') return result + + try { + raiseChatDbCompatibilityGate(getPathProvider(), { + version: getDesktopAppVersion(app.getVersion()), + kind: 'desktop', + }) + } catch (error) { + deleteImportedSessionFiles(result.sessionId) + return { + success: false, + error: error instanceof Error ? error.message : String(error), + diagnostics: result.diagnostics, + } + } + + return result }) } catch (error) { - deleteImportedSessionFiles(result.sessionId) - return { - success: false, - error: error instanceof Error ? error.message : String(error), - diagnostics: result.diagnostics, + if (error instanceof ImportInProgressError) { + return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY } } + throw error } - - return result } function deleteImportedSessionFiles(sessionId: string): void { @@ -981,9 +1013,17 @@ export async function incrementalImport( onProgress?: (progress: ParseProgress) => void, options?: ImportOptions ): Promise { - assertDataDirCompatibleNow() - - return sendToWorkerWithProgress('incrementalImport', { sessionId, filePath, options }, onProgress) + try { + return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => { + assertDataDirCompatibleNow() + return sendToWorkerWithProgress('incrementalImport', { sessionId, filePath, options }, onProgress) + }) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { success: false, newMessageCount: 0, error: IMPORT_IN_PROGRESS_ERROR_KEY } + } + throw error + } } /** diff --git a/apps/desktop/preload/apis/chat.ts b/apps/desktop/preload/apis/chat.ts index 7d617c7b..5040c681 100644 --- a/apps/desktop/preload/apis/chat.ts +++ b/apps/desktop/preload/apis/chat.ts @@ -13,6 +13,7 @@ interface ChatImportResult { error?: string importMode?: 'created' | 'incremental' matchedBy?: 'source-session-id' | 'stable-id' | 'trailing-messages' + createReason?: 'no-match' | 'ambiguous' newMessageCount?: number duplicateCount?: number diagnostics?: { @@ -20,6 +21,7 @@ interface ChatImportResult { detectedFormat: string | null messagesReceived: number messagesWritten: number + duplicateCount: number messagesSkipped: number skipReasons: { noSenderId: number; noAccountName: number; invalidTimestamp: number; noType: number } } diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index b67b9c7d..3f6c1047 100644 --- a/apps/desktop/preload/index.d.ts +++ b/apps/desktop/preload/index.d.ts @@ -27,6 +27,8 @@ interface ImportDiagnostics { messagesReceived: number /** 写入的消息数 */ messagesWritten: number + /** 去重过滤的消息数 */ + duplicateCount: number /** 跳过的消息数 */ messagesSkipped: number /** 跳过原因统计 */ @@ -44,6 +46,7 @@ interface ChatImportResult { error?: string importMode?: 'created' | 'incremental' matchedBy?: 'source-session-id' | 'stable-id' | 'trailing-messages' + createReason?: 'no-match' | 'ambiguous' newMessageCount?: number duplicateCount?: number diagnostics?: ImportDiagnostics diff --git a/docs/cn/standard/chatlab-import.md b/docs/cn/standard/chatlab-import.md index a168435f..e06f8260 100644 --- a/docs/cn/standard/chatlab-import.md +++ b/docs/cn/standard/chatlab-import.md @@ -371,7 +371,7 @@ ChatLab 不为调用方维护游标。推荐结构： ### 并发约束 -当前版本：**同一 session 同一时刻仅允许一个导入任务**。对同一 sessionId 的并发请求会收到 `IMPORT_IN_PROGRESS` (409) 错误；不同 session 的并发导入互不影响。 +当前版本：**同一用户数据目录同一时刻仅允许一个写入型导入任务**。如果 Desktop、CLI、Web 或 Push API 已在向该数据目录导入，后续请求即使目标 sessionId 不同，也会收到 `IMPORT_IN_PROGRESS` (409) 错误。格式检测、导入分析等只读操作不受此限制。 --- diff --git a/docs/en/standard/chatlab-import.md b/docs/en/standard/chatlab-import.md index 2d14973a..8eac4729 100644 --- a/docs/en/standard/chatlab-import.md +++ b/docs/en/standard/chatlab-import.md @@ -371,7 +371,7 @@ After each successful batch, update the cursor with `session.lastTimestamp` from ### Concurrency -Current version: **only one import task is allowed at a time per session**. Concurrent requests to the same sessionId return `IMPORT_IN_PROGRESS` (409); imports to different sessions run independently. +Current version: **only one write import task is allowed at a time per user data directory**. If Desktop, CLI, Web, or the Push API is already importing into that data directory, later requests return `IMPORT_IN_PROGRESS` (409) even when they target a different sessionId. Read-only format detection and import analysis are not blocked. --- diff --git a/packages/http-routes/src/errors.test.ts b/packages/http-routes/src/errors.test.ts index cc5dfd5d..9d54538d 100644 --- a/packages/http-routes/src/errors.test.ts +++ b/packages/http-routes/src/errors.test.ts @@ -2,6 +2,14 @@ import assert from 'node:assert/strict' import test from 'node:test' import { DataDirCompatibilityError } from '@openchatlab/node-runtime/src/data-dir-compat' import { ApiErrorCode, apiErrorFromUnknown } from './errors' +import { ImportInProgressError } from '@openchatlab/node-runtime/src/import/import-lock' + +test('apiErrorFromUnknown maps the shared import lock error to 409', () => { + const apiError = apiErrorFromUnknown(new ImportInProgressError()) + + assert.equal(apiError?.code, ApiErrorCode.IMPORT_IN_PROGRESS) + assert.equal(apiError?.statusCode, 409) +}) test('apiErrorFromUnknown maps data directory compatibility errors to 409', () => { const apiError = apiErrorFromUnknown( diff --git a/packages/http-routes/src/errors.ts b/packages/http-routes/src/errors.ts index ecce6e55..2b292325 100644 --- a/packages/http-routes/src/errors.ts +++ b/packages/http-routes/src/errors.ts @@ -5,6 +5,7 @@ */ import { DataDirCompatibilityError } from '@openchatlab/node-runtime/src/data-dir-compat' +import { ImportInProgressError } from '@openchatlab/node-runtime/src/import/import-lock' export enum ApiErrorCode { UNAUTHORIZED = 'UNAUTHORIZED', @@ -93,6 +94,7 @@ export function dataDirIncompatible(message: string): ApiError { export function apiErrorFromUnknown(error: unknown): ApiError | null { if (error instanceof ApiError) return error + if (error instanceof ImportInProgressError) return importInProgress() if (error instanceof DataDirCompatibilityError) return dataDirIncompatible(error.message) return null } diff --git a/packages/node-runtime/src/database-manager.ts b/packages/node-runtime/src/database-manager.ts index 686ff5d8..f9c65419 100644 --- a/packages/node-runtime/src/database-manager.ts +++ b/packages/node-runtime/src/database-manager.ts @@ -198,6 +198,10 @@ export class DatabaseManager { }) } + getUserDataDir(): string { + return this.pathProvider.getUserDataDir() + } + /** * 获取数据库文件路径 */ 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 c6fcaea9..9600810f 100644 --- a/packages/node-runtime/src/import/auto-import-matcher.test.ts +++ b/packages/node-runtime/src/import/auto-import-matcher.test.ts @@ -172,6 +172,114 @@ test('matches a unique private session by stable owner and participant ids', asy ) }) +test('falls back to trailing messages when private stable identity has drifted', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const sourceMeta: SourceMeta = { + name: 'Private chat', + platform: 'line', + type: 'private', + ownerId: 'owner', + } + const candidateMembers = [ + { platformId: 'owner', accountName: 'Owner' }, + { platformId: 'peer', accountName: 'Peer' }, + ] + const sourceMembers = [...candidateMembers, { platformId: 'new-device-member', accountName: 'Peer' }] + const messages = [ + textMessage('owner', 1783840151, 'one'), + textMessage('peer', 1783840152, 'two'), + textMessage('owner', 1783840153, 'three'), + textMessage('peer', 1783840154, 'four'), + textMessage('owner', 1783840155, 'five'), + ] + + seedSession(path.join(tempDir, 'existing.db'), sourceMeta, candidateMembers, messages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, sourceMeta, sourceMembers, messages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['existing'])), { + action: 'incremental', + sessionId: 'existing', + matchedBy: 'trailing-messages', + }) +}) + +test('prefers a unique stable identity over a different trailing-message candidate', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const sourceMeta: SourceMeta = { + name: 'Private chat', + platform: 'line', + type: 'private', + ownerId: 'owner', + } + const sourceMembers = [ + { platformId: 'owner', accountName: 'Owner' }, + { platformId: 'peer', accountName: 'Peer' }, + ] + const messages = [ + textMessage('owner', 1783840161, 'one'), + textMessage('peer', 1783840162, 'two'), + textMessage('owner', 1783840163, 'three'), + textMessage('peer', 1783840164, 'four'), + textMessage('owner', 1783840165, 'five'), + ] + + seedSession(path.join(tempDir, 'stable.db'), sourceMeta, sourceMembers, [textMessage('peer', 1, 'old')]) + seedSession(path.join(tempDir, 'trailing.db'), { ...sourceMeta, ownerId: 'different-owner' }, sourceMembers, messages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, sourceMeta, sourceMembers, messages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['stable', 'trailing'])), { + action: 'incremental', + sessionId: 'stable', + matchedBy: 'stable-id', + }) +}) + +test('uses a unique trailing overlap to disambiguate multiple stable matches', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { + name: 'Renamed group', + platform: 'qq', + type: 'group', + groupId: 'group-42', + } + const members = [{ platformId: 'member', accountName: 'Member' }] + const messages = [ + textMessage('member', 1783840171, 'one'), + textMessage('member', 1783840172, 'two'), + textMessage('member', 1783840173, 'three'), + textMessage('member', 1783840174, 'four'), + textMessage('member', 1783840175, 'five'), + ] + + seedSession(path.join(tempDir, 'matching-tail.db'), meta, members, messages) + seedSession(path.join(tempDir, 'different-tail.db'), meta, members, [ + textMessage('member', 1, 'other one'), + textMessage('member', 2, 'other two'), + textMessage('member', 3, 'other three'), + textMessage('member', 4, 'other four'), + textMessage('member', 5, 'other five'), + ]) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, members, messages) + + assert.deepEqual( + await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['matching-tail', 'different-tail'])), + { + action: 'incremental', + sessionId: 'matching-tail', + matchedBy: 'trailing-messages', + } + ) +}) + test('uses a validated ChatLab source session id to disambiguate stable matches', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -395,3 +503,54 @@ test('does not match a candidate with fewer than five business messages', async reason: 'no-match', }) }) + +test('ignores non-chat databases while inspecting Desktop-style candidates', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { name: 'LINE chat', platform: 'line', type: 'private' } + const members = [{ platformId: 'Alice', accountName: 'Alice' }] + const messages = [ + textMessage('Alice', 1783840701, 'one'), + textMessage('Alice', 1783840702, 'two'), + textMessage('Alice', 1783840703, 'three'), + textMessage('Alice', 1783840704, 'four'), + textMessage('Alice', 1783840705, 'five'), + ] + seedSession(path.join(tempDir, 'existing.db'), meta, members, messages) + const unrelatedDb = openBetterSqliteDatabase(path.join(tempDir, 'unrelated.db'), { nativeBinding }) + unrelatedDb.exec('CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT)') + unrelatedDb.close() + + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, members, messages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['unrelated', 'existing'])), { + action: 'incremental', + sessionId: 'existing', + matchedBy: 'trailing-messages', + }) +}) + +test('reports the candidate session id when a database cannot be inspected', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl( + sourcePath, + { name: 'LINE chat', platform: 'line', type: 'private' }, + [{ platformId: 'Alice', accountName: 'Alice' }], + [textMessage('Alice', 1783840801, 'hello')] + ) + + await assert.rejects( + resolveAutoImportTarget(sourcePath, { + listSessionIds: () => ['broken-session'], + openReadonly: () => { + throw new Error('database disk image is malformed') + }, + }), + /broken-session.*database disk image is malformed/ + ) +}) diff --git a/packages/node-runtime/src/import/auto-import-matcher.ts b/packages/node-runtime/src/import/auto-import-matcher.ts index b92aef0c..e380e2ea 100644 --- a/packages/node-runtime/src/import/auto-import-matcher.ts +++ b/packages/node-runtime/src/import/auto-import-matcher.ts @@ -1,4 +1,4 @@ -import { generateMessageKey, getSessionMeta, type DatabaseAdapter } from '@openchatlab/core' +import { generateMessageKey, getSessionMeta, isChatSessionDb, type DatabaseAdapter } from '@openchatlab/core' import { streamParseFile, type ParsedMeta } from '@openchatlab/parser' import { MessageType } from '@openchatlab/shared-types' import { normalizeImportTimestamp } from './incremental-importer' @@ -7,10 +7,11 @@ import type { ImportProgressCallback } from './streaming-importer' const MATCH_WINDOW_SIZE = 5 export type AutoImportMatchMethod = 'source-session-id' | 'stable-id' | 'trailing-messages' +export type AutoImportCreateReason = 'no-match' | 'ambiguous' export type AutoImportDecision = | { action: 'incremental'; sessionId: string; matchedBy: AutoImportMatchMethod } - | { action: 'create'; reason: 'no-match' | 'ambiguous' } + | { action: 'create'; reason: AutoImportCreateReason } export interface AutoImportMatcherDeps { listSessionIds(): string[] @@ -83,8 +84,12 @@ export async function resolveAutoImportTarget( const stableMatches: string[] = [] const trailingMatches: string[] = [] for (const sessionId of deps.listSessionIds()) { - const db = deps.openReadonly(sessionId) + let db: DatabaseAdapter | null = null try { + db = deps.openReadonly(sessionId) + // Desktop 的数据库目录还可能包含配置库等非聊天 DB，匹配前统一排除。 + if (!isChatSessionDb(db)) continue + const candidate = getSessionMeta(db) if (candidate?.platform !== meta.platform || candidate.type !== meta.type) continue @@ -105,7 +110,7 @@ export async function resolveAutoImportTarget( stableMatches.push(sessionId) } - if (!hasStableIdentity && sourceWindows.size > 0) { + if (sourceWindows.size > 0) { const rows = db .prepare( `SELECT msg.ts, member.platform_id, msg.type, msg.content @@ -131,8 +136,13 @@ export async function resolveAutoImportTarget( if (sourceWindows.has(signature)) trailingMatches.push(sessionId) } } + } catch (error) { + const detail = error instanceof Error ? error.message : String(error) + throw new Error(`Failed to inspect import candidate ${JSON.stringify(sessionId)}: ${detail}`, { + cause: error, + }) } finally { - db.close() + db?.close() } } @@ -143,13 +153,15 @@ export async function resolveAutoImportTarget( if (stableMatches.length === 1) { return { action: 'incremental', sessionId: stableMatches[0], matchedBy: 'stable-id' } } - return { action: 'create', reason: stableMatches.length > 1 ? 'ambiguous' : 'no-match' } } + + // 稳定身份缺失、漂移或产生多个候选时，仍只接受唯一的 5 条连续消息重叠。 if (meta.sourceSessionId && trailingMatches.includes(meta.sourceSessionId)) { return { action: 'incremental', sessionId: meta.sourceSessionId, matchedBy: 'source-session-id' } } if (trailingMatches.length === 1) { return { action: 'incremental', sessionId: trailingMatches[0], matchedBy: 'trailing-messages' } } - return { action: 'create', reason: trailingMatches.length > 1 ? 'ambiguous' : 'no-match' } + const ambiguous = stableMatches.length > 1 || trailingMatches.length > 1 + return { action: 'create', reason: ambiguous ? 'ambiguous' : 'no-match' } } diff --git a/packages/node-runtime/src/import/auto-importer.test.ts b/packages/node-runtime/src/import/auto-importer.test.ts index d7a611a2..5160f650 100644 --- a/packages/node-runtime/src/import/auto-importer.test.ts +++ b/packages/node-runtime/src/import/auto-importer.test.ts @@ -39,6 +39,7 @@ function createDeps(options?: { detectedFormat: 'fixture', messagesReceived: 12, messagesWritten: 10, + duplicateCount: 0, messagesSkipped: 2, skipReasons: { noSenderId: 0, noAccountName: 0, invalidTimestamp: 2, noType: 0 }, }, @@ -125,6 +126,7 @@ for (const reason of ['no-match', 'ambiguous'] as const) { assert.equal(result.success, true) assert.equal(result.importMode, 'created') + assert.equal(result.createReason, reason) assert.equal(calls.match, 1) assert.equal(calls.create.length, 1) assert.deepEqual(calls.append, []) diff --git a/packages/node-runtime/src/import/auto-importer.ts b/packages/node-runtime/src/import/auto-importer.ts index e6213a31..d03a68f5 100644 --- a/packages/node-runtime/src/import/auto-importer.ts +++ b/packages/node-runtime/src/import/auto-importer.ts @@ -2,6 +2,7 @@ import { appLogger } from '../logging/app-logger' import { resolveAutoImportTarget, type AutoImportDecision, + type AutoImportCreateReason, type AutoImportMatcherDeps, type AutoImportMatchMethod, } from './auto-import-matcher' @@ -33,13 +34,14 @@ export interface AutoImportResult { sessionId?: string importMode?: 'created' | 'incremental' matchedBy?: AutoImportMatchMethod + createReason?: AutoImportCreateReason newMessageCount?: number duplicateCount?: number diagnostics?: ImportDiagnostics error?: string } -function mapCreateResult(result: StreamImportResult): AutoImportResult { +function mapCreateResult(result: StreamImportResult, createReason?: AutoImportCreateReason): AutoImportResult { if (!result.success || !result.sessionId) { return { success: false, error: result.error, diagnostics: result.diagnostics } } @@ -47,8 +49,9 @@ function mapCreateResult(result: StreamImportResult): AutoImportResult { success: true, sessionId: result.sessionId, importMode: 'created', + ...(createReason ? { createReason } : {}), newMessageCount: result.diagnostics?.messagesWritten ?? 0, - duplicateCount: 0, + duplicateCount: result.diagnostics?.duplicateCount ?? 0, diagnostics: result.diagnostics, } } @@ -126,7 +129,7 @@ export async function autoImportFile( return result } - const result = mapCreateResult(await deps.createSession(filePath, options.formatOptions)) + const result = mapCreateResult(await deps.createSession(filePath, options.formatOptions), decision.reason) appLogger.info('import', 'Automatic import created a new session', { reason: decision.reason, sessionId: result.sessionId, diff --git a/packages/node-runtime/src/import/import-lock.test.ts b/packages/node-runtime/src/import/import-lock.test.ts new file mode 100644 index 00000000..b17318cc --- /dev/null +++ b/packages/node-runtime/src/import/import-lock.test.ts @@ -0,0 +1,82 @@ +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 { ImportInProgressError, withDataDirImportLock } from './import-lock' + +function makeTempDir(): string { + const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() + return fs.mkdtempSync(path.join(baseDir, 'chatlab-import-lock-')) +} + +test('rejects a second import for the same data directory and releases after completion', async (t) => { + const dataDir = makeTempDir() + t.after(() => fs.rmSync(dataDir, { recursive: true, force: true })) + + let releaseFirst!: () => void + const firstCanFinish = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstDidStart = new Promise((resolve) => { + firstStarted = resolve + }) + + const first = withDataDirImportLock(dataDir, async () => { + firstStarted() + await firstCanFinish + return 'first' + }) + await firstDidStart + + await assert.rejects( + () => withDataDirImportLock(dataDir, async () => 'second'), + (error: unknown) => error instanceof ImportInProgressError && error.code === 'IMPORT_IN_PROGRESS' + ) + + releaseFirst() + assert.equal(await first, 'first') + assert.equal(await withDataDirImportLock(dataDir, async () => 'third'), 'third') +}) + +test('allows imports in different data directories', async (t) => { + const firstDir = makeTempDir() + const secondDir = makeTempDir() + t.after(() => { + fs.rmSync(firstDir, { recursive: true, force: true }) + fs.rmSync(secondDir, { recursive: true, force: true }) + }) + + let releaseFirst!: () => void + const firstCanFinish = new Promise((resolve) => { + releaseFirst = resolve + }) + let firstStarted!: () => void + const firstDidStart = new Promise((resolve) => { + firstStarted = resolve + }) + + const first = withDataDirImportLock(firstDir, async () => { + firstStarted() + await firstCanFinish + }) + await firstDidStart + + assert.equal(await withDataDirImportLock(secondDir, async () => 'second'), 'second') + releaseFirst() + await first +}) + +test('recovers a lock left by a process that is no longer running', async (t) => { + const dataDir = makeTempDir() + t.after(() => fs.rmSync(dataDir, { recursive: true, force: true })) + + fs.writeFileSync( + path.join(dataDir, '.chatlab-import.lock'), + JSON.stringify({ pid: 2_147_483_647, token: 'stale', startedAt: 1 }), + 'utf-8' + ) + + assert.equal(await withDataDirImportLock(dataDir, async () => 'recovered'), 'recovered') +}) diff --git a/packages/node-runtime/src/import/import-lock.ts b/packages/node-runtime/src/import/import-lock.ts new file mode 100644 index 00000000..08faaa6f --- /dev/null +++ b/packages/node-runtime/src/import/import-lock.ts @@ -0,0 +1,131 @@ +import * as fs from 'node:fs' +import * as path from 'node:path' +import { randomUUID } from 'node:crypto' +import { appLogger } from '../logging/app-logger' + +export const IMPORT_LOCK_FILENAME = '.chatlab-import.lock' +export const IMPORT_IN_PROGRESS_ERROR_KEY = 'error.import_in_progress' + +interface ImportLockOwner { + pid: number + token: string + startedAt: number +} + +export class ImportInProgressError extends Error { + readonly code = 'IMPORT_IN_PROGRESS' + readonly errorKey = IMPORT_IN_PROGRESS_ERROR_KEY + + constructor() { + super('Another import is already in progress') + this.name = 'ImportInProgressError' + } +} + +function parseOwner(lockPath: string): ImportLockOwner | null { + try { + const parsed = JSON.parse(fs.readFileSync(lockPath, 'utf-8')) as Partial + if ( + typeof parsed.pid !== 'number' || + !Number.isInteger(parsed.pid) || + parsed.pid <= 0 || + typeof parsed.token !== 'string' || + parsed.token.length === 0 || + typeof parsed.startedAt !== 'number' + ) { + return null + } + return parsed as ImportLockOwner + } catch { + return null + } +} + +function isProcessRunning(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch (error) { + return (error as NodeJS.ErrnoException).code !== 'ESRCH' + } +} + +function removeStaleLock(lockPath: string): boolean { + const owner = parseOwner(lockPath) + if (owner) { + if (isProcessRunning(owner.pid)) return false + } else { + try { + // A competing process may have created the file but not written its owner yet. + // Only treat malformed lock files as stale after that short creation window. + if (Date.now() - fs.statSync(lockPath).mtimeMs < 5_000) return false + } catch { + return true + } + } + + try { + fs.unlinkSync(lockPath) + appLogger.warn('import-lock', 'Removed stale import lock', { ownerPid: owner?.pid }) + return true + } catch { + return false + } +} + +function releaseLock(lockPath: string, fd: number, owner: ImportLockOwner): void { + try { + fs.closeSync(fd) + } catch (error) { + appLogger.error('import-lock', 'Failed to close import lock', error) + } + + try { + if (parseOwner(lockPath)?.token !== owner.token) { + appLogger.warn('import-lock', 'Import lock ownership changed before release') + return + } + fs.unlinkSync(lockPath) + appLogger.info('import-lock', 'Import lock released') + } catch (error) { + appLogger.error('import-lock', 'Failed to release import lock', error) + } +} + +export async function withDataDirImportLock(userDataDir: string, task: () => Promise): Promise { + fs.mkdirSync(userDataDir, { recursive: true }) + const lockPath = path.join(userDataDir, IMPORT_LOCK_FILENAME) + + let fd: number | null = null + for (let attempt = 0; attempt < 2; attempt++) { + try { + fd = fs.openSync(lockPath, 'wx') + break + } catch (error) { + const code = (error as NodeJS.ErrnoException).code + if (code === 'EEXIST' && attempt === 0 && removeStaleLock(lockPath)) continue + if (code === 'EEXIST') { + appLogger.warn('import-lock', 'Rejected concurrent import') + throw new ImportInProgressError() + } + throw error + } + } + + if (fd === null) throw new ImportInProgressError() + + const owner: ImportLockOwner = { + pid: process.pid, + token: randomUUID(), + startedAt: Date.now(), + } + + try { + fs.writeFileSync(fd, JSON.stringify(owner), 'utf-8') + fs.fsyncSync(fd) + appLogger.info('import-lock', 'Import lock acquired') + return await task() + } finally { + releaseLock(lockPath, fd, owner) + } +} diff --git a/packages/node-runtime/src/import/incremental-importer.test.ts b/packages/node-runtime/src/import/incremental-importer.test.ts index 8ec73326..aa8064ab 100644 --- a/packages/node-runtime/src/import/incremental-importer.test.ts +++ b/packages/node-runtime/src/import/incremental-importer.test.ts @@ -131,3 +131,48 @@ test('preserves ChatLab JSON member aliases during incremental import', async (t assert.deepEqual(JSON.parse(row?.aliases ?? '[]'), ['Ally']) }) + +test('does not deduplicate messages that only share timestamp, sender and content', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const dbPath = path.join(tempDir, 'session.db') + const filePath = path.join(tempDir, 'different-type.json') + seedSessionDb(dbPath) + + const db = openBetterSqliteDatabase(dbPath, { nativeBinding }) + db.prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)').run('wxid_alice', 'Alice') + const member = db.prepare('SELECT id FROM member WHERE platform_id = ?').get('wxid_alice') as { id: number } + db.prepare('INSERT INTO message (sender_id, ts, type, content) VALUES (?, ?, ?, ?)').run( + member.id, + 1780330832, + 0, + 'same content' + ) + db.close() + + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Different Type', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 1, + content: 'same content', + }, + ], + }), + 'utf8' + ) + + const result = await incrementalImport('session', filePath, createDeps(dbPath)) + + assert.equal(result.success, true) + assert.equal(result.newMessageCount, 1) + assert.equal(result.batch?.duplicateCount, 0) +}) diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index d258f083..aea428fc 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -9,9 +9,10 @@ */ import type { DatabaseAdapter } from '@openchatlab/core' -import { generateMessageKey, generateSessionIndex, generateIncrementalSessionIndex } from '@openchatlab/core' +import { generateSessionIndex, generateIncrementalSessionIndex } from '@openchatlab/core' import { streamParseFile, detectFormat, getFormatFeatureById, type ParseProgress } from '@openchatlab/parser' import { insertFtsEntries, hasFtsTable } from '../fts' +import { generateFallbackMessageKey, registerMessageAndCheckDuplicate } from './message-deduplicator' import type { ImportProgressCallback } from './streaming-importer' // ==================== Public interfaces ==================== @@ -87,32 +88,48 @@ function loadExistingDedup(db: DatabaseAdapter): { const hashRows = db .prepare( - `SELECT ts, m.platform_id as sender_platform_id, content + `SELECT ts, m.platform_id as sender_platform_id, msg.type, content, reply_to_message_id FROM message msg JOIN member m ON msg.sender_id = m.id` ) - .all() as Array<{ ts: number; sender_platform_id: string; content: string | null }> + .all() as Array<{ + ts: number + sender_platform_id: string + type: number + content: string | null + reply_to_message_id: string | null + }> for (const row of hashRows) { - existingKeys.add(generateMessageKey(row.ts, row.sender_platform_id, row.content)) + existingKeys.add( + generateFallbackMessageKey({ + timestamp: row.ts, + senderPlatformId: row.sender_platform_id, + type: row.type, + content: row.content, + replyToMessageId: row.reply_to_message_id ?? undefined, + }) + ) } return { existingPlatformMsgIds, existingKeys } } function isDuplicate( - msg: { platformMessageId?: string; timestamp: number; senderPlatformId: string; content: string | null }, + msg: { + platformMessageId?: string + timestamp: number + senderPlatformId: string + type: number + content: string | null + replyToMessageId?: string + }, existingPlatformMsgIds: Set, existingKeys: Set ): boolean { - if (msg.platformMessageId) { - if (existingPlatformMsgIds.has(msg.platformMessageId)) return true - existingPlatformMsgIds.add(msg.platformMessageId) - return false - } - const key = generateMessageKey(msg.timestamp, msg.senderPlatformId, msg.content) - if (existingKeys.has(key)) return true - existingKeys.add(key) - return false + return registerMessageAndCheckDuplicate(msg, { + platformMessageIds: existingPlatformMsgIds, + fallbackKeys: existingKeys, + }) } export function normalizeImportTimestamp(timestamp: unknown): number | null { diff --git a/packages/node-runtime/src/import/index.ts b/packages/node-runtime/src/import/index.ts index 39c4eb7f..7ed2d438 100644 --- a/packages/node-runtime/src/import/index.ts +++ b/packages/node-runtime/src/import/index.ts @@ -33,9 +33,20 @@ export type { IncrementalImportDeps, } from './incremental-importer' export { resolveAutoImportTarget } from './auto-import-matcher' -export type { AutoImportDecision, AutoImportMatcherDeps, AutoImportMatchMethod } from './auto-import-matcher' +export type { + AutoImportCreateReason, + AutoImportDecision, + AutoImportMatcherDeps, + AutoImportMatchMethod, +} from './auto-import-matcher' export { autoImportFile } from './auto-importer' export type { AutoImportDeps, AutoImportOptions, AutoImportResult } from './auto-importer' +export { + IMPORT_IN_PROGRESS_ERROR_KEY, + IMPORT_LOCK_FILENAME, + ImportInProgressError, + withDataDirImportLock, +} from './import-lock' export { ZipArchiveReader, validateArchiveEntryName } from './archive/archive-reader' export { ArchiveImportError } from './archive/errors' export { GoogleChatTakeoutResolver } from './archive/google-chat-resolver' diff --git a/packages/node-runtime/src/import/message-deduplicator.ts b/packages/node-runtime/src/import/message-deduplicator.ts new file mode 100644 index 00000000..41cb67a7 --- /dev/null +++ b/packages/node-runtime/src/import/message-deduplicator.ts @@ -0,0 +1,47 @@ +import { generateMessageKey } from '@openchatlab/core' + +export interface MessageDedupState { + platformMessageIds: Set + fallbackKeys: Set +} + +export interface DedupMessage { + platformMessageId?: string + timestamp: number + senderPlatformId: string + type: number + content: string | null + replyToMessageId?: string +} + +export function generateFallbackMessageKey(message: Omit): string { + const contentKey = generateMessageKey(message.timestamp, message.senderPlatformId, message.content) + return JSON.stringify([contentKey, message.type, message.replyToMessageId || null]) +} + +export function createMessageDedupState( + platformMessageIds: Iterable = [], + fallbackKeys: Iterable = [] +): MessageDedupState { + return { + platformMessageIds: new Set(platformMessageIds), + fallbackKeys: new Set(fallbackKeys), + } +} + +/** + * Canonical file-import dedup rule: prefer stable platform message IDs and + * fall back to timestamp + sender platform ID + type + normalized content + reply target. + */ +export function registerMessageAndCheckDuplicate(message: DedupMessage, state: MessageDedupState): boolean { + if (message.platformMessageId) { + if (state.platformMessageIds.has(message.platformMessageId)) return true + state.platformMessageIds.add(message.platformMessageId) + return false + } + + const key = generateFallbackMessageKey(message) + if (state.fallbackKeys.has(key)) return true + state.fallbackKeys.add(key) + return false +} diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index b194d5b3..bb92fe73 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/packages/node-runtime/src/import/streaming-importer.test.ts @@ -89,6 +89,44 @@ function writeChunkedQqExport(root: string): string { return manifestPath } +function writeDuplicateChatLabExport(root: string): string { + const filePath = path.join(root, 'duplicate-chat.json') + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 1711468800 }, + meta: { name: 'Duplicate Test', platform: 'instagram', type: 'private', ownerId: 'owner' }, + members: [ + { platformId: 'owner', accountName: 'Owner' }, + { platformId: 'friend', accountName: 'Friend' }, + ], + messages: [ + { sender: 'friend', accountName: 'Friend', timestamp: 1711468800, type: 0, content: 'same message' }, + { sender: 'friend', accountName: 'Friend', timestamp: 1711468800, type: 0, content: 'same message' }, + { sender: 'friend', accountName: 'Friend', timestamp: 1711468800, type: 1, content: 'same message' }, + { + sender: 'owner', + accountName: 'Owner', + timestamp: 1711468801, + type: 0, + content: 'message with id', + platformMessageId: 'platform-1', + }, + { + sender: 'owner', + accountName: 'Owner', + timestamp: 1711468801, + type: 0, + content: 'message with id', + platformMessageId: 'platform-1', + }, + ], + }), + 'utf-8' + ) + return filePath +} + test('streamingImport updates avatars for members first created from message batches', async () => { const root = makeTempDir() const manifestPath = writeChunkedQqExport(root) @@ -130,3 +168,45 @@ test('streamingImport updates avatars for members first created from message bat assert.equal(row?.platform_id, '10001') assert.equal(row?.avatar, 'data:image/png;base64,AAAA') }) + +test('streamingImport applies incremental-equivalent deduplication on first import', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + const filePath = writeDuplicateChatLabExport(root) + const dbPath = path.join(root, 'duplicate-test.db') + + const result = await streamingImport( + filePath, + { + openDatabase() { + const db = new Database(dbPath, { nativeBinding }) + db.exec(CHAT_DB_TABLES) + return new BetterSqliteAdapter(db) + }, + deleteDatabase() { + for (const suffix of ['', '-wal', '-shm']) { + try { + fs.unlinkSync(dbPath + suffix) + } catch { + /* ignore */ + } + } + }, + onProgress() { + /* noop for this focused importer test */ + }, + }, + undefined, + 'duplicate-test' + ) + + assert.equal(result.success, true) + assert.equal(result.diagnostics?.messagesReceived, 5) + assert.equal(result.diagnostics?.messagesWritten, 3) + assert.equal(result.diagnostics?.duplicateCount, 2) + + const db = new Database(dbPath, { readonly: true, nativeBinding }) + const row = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number } + db.close() + assert.equal(row.count, 3) +}) diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index d6c10101..74432d83 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -25,6 +25,7 @@ import { } from '@openchatlab/parser' import * as fs from 'fs' import { buildFtsIndex } from '../fts' +import { createMessageDedupState, registerMessageAndCheckDuplicate } from './message-deduplicator' // ==================== Public interfaces ==================== @@ -40,6 +41,7 @@ export interface ImportDiagnostics { detectedFormat: string | null messagesReceived: number messagesWritten: number + duplicateCount: number messagesSkipped: number skipReasons: SkipReasons } @@ -245,6 +247,7 @@ async function streamImportSingle( let metaInserted = false let messageCountInBatch = 0 let totalMessageCount = 0 + let duplicateCount = 0 let lastCheckpointCount = 0 let inTransaction = false @@ -305,6 +308,7 @@ async function streamImportSingle( skippedInvalidTimestamp: 0, skippedNoType: 0, } + const dedupState = createMessageDedupState() logger?.info('Starting streamParseFile...') @@ -397,6 +401,28 @@ async function streamImportSingle( continue } + let safeContent: string | null = null + if (msg.content != null) { + safeContent = typeof msg.content === 'string' ? msg.content : JSON.stringify(msg.content) + } + + if ( + registerMessageAndCheckDuplicate( + { + platformMessageId: msg.platformMessageId, + timestamp: msg.timestamp, + senderPlatformId: msg.senderPlatformId, + type: msg.type, + content: safeContent, + replyToMessageId: msg.replyToMessageId, + }, + dedupState + ) + ) { + duplicateCount++ + continue + } + let t0 = Date.now() if (!memberIdMap.has(msg.senderPlatformId)) { insertMember.run( @@ -419,11 +445,6 @@ async function streamImportSingle( const senderId = memberIdMap.get(msg.senderPlatformId) if (senderId === undefined) continue - let safeContent: string | null = null - if (msg.content != null) { - safeContent = typeof msg.content === 'string' ? msg.content : JSON.stringify(msg.content) - } - t0 = Date.now() insertMessage.run( senderId, @@ -581,6 +602,7 @@ async function streamImportSingle( logger?.info(` invalid timestamp: ${callbackStats.skippedInvalidTimestamp}`) if (callbackStats.skippedNoType > 0) logger?.info(` missing type: ${callbackStats.skippedNoType}`) } + if (duplicateCount > 0) logger?.info(`Duplicate messages skipped: ${duplicateCount}`) logger?.summary(totalMessageCount, memberIdMap.size) @@ -619,6 +641,7 @@ async function streamImportSingle( detectedFormat: formatFeature ? `${formatFeature.name} (${formatFeature.id})` : null, messagesReceived: callbackStats.totalMessagesReceived, messagesWritten: totalMessageCount, + duplicateCount, messagesSkipped: callbackStats.skippedNoSenderId + callbackStats.skippedNoAccountName + diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 29c34ff2..04dda5ef 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -17,6 +17,10 @@ export { incrementalImport, resolveAutoImportTarget, autoImportFile, + IMPORT_IN_PROGRESS_ERROR_KEY, + IMPORT_LOCK_FILENAME, + ImportInProgressError, + withDataDirImportLock, logNativeParserStatus, ZipArchiveReader, validateArchiveEntryName, @@ -41,6 +45,7 @@ export type { IncrementalImportResult, IncrementalImportDeps, AutoImportDecision, + AutoImportCreateReason, AutoImportMatcherDeps, AutoImportMatchMethod, AutoImportDeps, diff --git a/packages/node-runtime/src/services/push-importer.test.ts b/packages/node-runtime/src/services/push-importer.test.ts index fe03ca2a..d074819c 100644 --- a/packages/node-runtime/src/services/push-importer.test.ts +++ b/packages/node-runtime/src/services/push-importer.test.ts @@ -4,6 +4,7 @@ import os from 'node:os' import path from 'node:path' import test from 'node:test' import { DatabaseManager } from '../database-manager' +import { withDataDirImportLock } from '../import/import-lock' import type { PushImportPayload } from './push-importer' import { pushImport } from './push-importer' @@ -32,6 +33,27 @@ function createDatabaseManager(rootDir: string): DatabaseManager { ) } +test('rejects push imports while any writer holds the data-directory import lock', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + const manager = createDatabaseManager(root) + const payload: PushImportPayload = { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Concurrent Session', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }], + } + + const outcome = await withDataDirImportLock(root, () => pushImport(manager, 'different-session', payload)) + + assert.deepEqual(outcome, { + ok: false, + reason: 'import_in_progress', + message: 'Another import is already in progress', + }) + assert.equal(fs.existsSync(manager.getDbPath('different-session')), false) +}) + test('deduplicates initial push batch when a platform-id message is repeated without platform id', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) diff --git a/packages/node-runtime/src/services/push-importer.ts b/packages/node-runtime/src/services/push-importer.ts index c775ef35..ee2c36d8 100644 --- a/packages/node-runtime/src/services/push-importer.ts +++ b/packages/node-runtime/src/services/push-importer.ts @@ -17,9 +17,8 @@ import type { DatabaseAdapter } from '@openchatlab/core' import type { DatabaseManager } from '../database-manager' import { buildFtsIndex, insertFtsEntries } from '../fts' import { writeParseResultToDb } from '../import' +import { ImportInProgressError, withDataDirImportLock } from '../import/import-lock' -// Per-session lock: concurrent imports to different sessions are fine (separate DB files). -const importInProgress = new Set() const SYSTEM_SENDER_ID = 'SYSTEM' const SYSTEM_MEMBER_NAME = '系统消息' @@ -437,14 +436,27 @@ export async function pushImport( return { ok: false, reason: 'invalid_payload', message: 'sessionId contains invalid characters' } } - if (importInProgress.has(sessionId)) { - return { - ok: false, - reason: 'import_in_progress', - message: 'Another import is already in progress for this session', + try { + return await withDataDirImportLock(dbManager.getUserDataDir(), () => + pushImportUnlocked(dbManager, sessionId, payload) + ) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { + ok: false, + reason: 'import_in_progress', + message: 'Another import is already in progress', + } } + throw error } +} +async function pushImportUnlocked( + dbManager: DatabaseManager, + sessionId: string, + payload: PushImportPayload +): Promise { const dbPath = dbManager.getDbPath(sessionId) const isNew = !fs.existsSync(dbPath) @@ -453,7 +465,6 @@ export async function pushImport( return { ok: false, reason: 'invalid_payload', message: validationError } } - importInProgress.add(sessionId) try { if (isNew) { const db = dbManager.openRawSessionDatabase(sessionId, { create: true }) @@ -512,7 +523,5 @@ export async function pushImport( } } return { ok: false, reason: 'import_failed', message: err instanceof Error ? err.message : String(err) } - } finally { - importInProgress.delete(sessionId) } } diff --git a/src/i18n/locales/en-US/home.json b/src/i18n/locales/en-US/home.json index 01c64081..11023756 100644 --- a/src/i18n/locales/en-US/home.json +++ b/src/i18n/locales/en-US/home.json @@ -33,6 +33,7 @@ "errors": { "no_file_selected": "No file selected", "import_failed": "Import failed", + "import_in_progress": "Another import is already in progress. Please try again later.", "unrecognized_format": "Unrecognized file format", "no_messages": "No messages parsed. Please check if the file format is correct.", "no_chat_selected": "No chat selected", @@ -56,6 +57,7 @@ }, "outcome": { "created": "New session created", + "ambiguous": "Multiple existing sessions matched, so they were not merged and a new session was created", "incremental": "Added {newCount} messages and skipped {duplicateCount} duplicates", "unchanged": "Already imported; no new messages" }, @@ -67,6 +69,7 @@ "skipped": "Skipped", "success": "Import successful", "created": "New session created", + "ambiguous": "Multiple candidates matched; created a new session", "incremental": "Added {newCount}; skipped {duplicateCount} duplicates", "unchanged": "Already imported; no new messages", "completed": "Batch import completed", diff --git a/src/i18n/locales/ja-JP/home.json b/src/i18n/locales/ja-JP/home.json index cc1d590d..55ad96c5 100644 --- a/src/i18n/locales/ja-JP/home.json +++ b/src/i18n/locales/ja-JP/home.json @@ -33,6 +33,7 @@ "errors": { "no_file_selected": "ファイルが選択されていません", "import_failed": "インポートに失敗しました", + "import_in_progress": "別のインポートが実行中です。しばらくしてから再試行してください。", "unrecognized_format": "認識できないファイル形式です", "no_messages": "メッセージを検出できませんでした。ファイル形式が正しいか確認してください", "no_chat_selected": "チャットが選択されていません", @@ -56,6 +57,7 @@ }, "outcome": { "created": "新しいセッションを作成しました", + "ambiguous": "既存セッションの候補が複数見つかったため、自動結合せず新しいセッションを作成しました", "incremental": "{newCount} 件を追加し、重複 {duplicateCount} 件をスキップしました", "unchanged": "インポート済みのため、新しいメッセージはありません" }, @@ -67,6 +69,7 @@ "skipped": "スキップ済み", "success": "インポート成功", "created": "新しいセッションを作成しました", + "ambiguous": "候補が複数見つかったため、新しいセッションを作成しました", "incremental": "{newCount} 件追加、重複 {duplicateCount} 件スキップ", "unchanged": "インポート済み、新規メッセージなし", "completed": "一括インポート完了", diff --git a/src/i18n/locales/zh-CN/home.json b/src/i18n/locales/zh-CN/home.json index 6c7ca863..ae784a9e 100644 --- a/src/i18n/locales/zh-CN/home.json +++ b/src/i18n/locales/zh-CN/home.json @@ -33,6 +33,7 @@ "errors": { "no_file_selected": "未选择文件", "import_failed": "导入失败", + "import_in_progress": "已有导入任务正在进行，请稍后重试", "unrecognized_format": "无法识别的文件格式", "no_messages": "未解析到任何消息，请检查文件格式是否正确", "no_chat_selected": "未选择聊天", @@ -56,6 +57,7 @@ }, "outcome": { "created": "已创建新会话", + "ambiguous": "发现多个可能的已有会话，未自动合并，已创建新会话", "incremental": "已增量导入 {newCount} 条消息，跳过 {duplicateCount} 条重复消息", "unchanged": "内容已导入，没有新增消息" }, @@ -67,6 +69,7 @@ "skipped": "已跳过", "success": "导入成功", "created": "已创建新会话", + "ambiguous": "候选会话不唯一，已创建新会话", "incremental": "新增 {newCount} 条，跳过 {duplicateCount} 条重复消息", "unchanged": "内容已导入，无新增消息", "completed": "批量导入完成", diff --git a/src/i18n/locales/zh-TW/home.json b/src/i18n/locales/zh-TW/home.json index e1e016fd..be2f2e86 100644 --- a/src/i18n/locales/zh-TW/home.json +++ b/src/i18n/locales/zh-TW/home.json @@ -33,6 +33,7 @@ "errors": { "no_file_selected": "未選擇檔案", "import_failed": "匯入失敗", + "import_in_progress": "已有匯入工作正在進行，請稍後再試", "unrecognized_format": "無法識別的檔案格式", "no_messages": "未解析到任何訊息，請檢查檔案格式是否正確", "no_chat_selected": "未選擇聊天", @@ -56,6 +57,7 @@ }, "outcome": { "created": "已建立新會話", + "ambiguous": "找到多個可能的既有會話，因此未自動合併並建立了新會話", "incremental": "已增量匯入 {newCount} 則訊息，略過 {duplicateCount} 則重複訊息", "unchanged": "內容已匯入，沒有新增訊息" }, @@ -67,6 +69,7 @@ "skipped": "已跳過", "success": "匯入成功", "created": "已建立新會話", + "ambiguous": "候選會話不唯一，已建立新會話", "incremental": "新增 {newCount} 則，略過 {duplicateCount} 則重複訊息", "unchanged": "內容已匯入，無新增訊息", "completed": "批次匯入完成", diff --git a/src/pages/home/components/import/ImportArea.vue b/src/pages/home/components/import/ImportArea.vue index 01d63a6e..d6522c82 100644 --- a/src/pages/home/components/import/ImportArea.vue +++ b/src/pages/home/components/import/ImportArea.vue @@ -53,7 +53,7 @@ const showFormatSelector = ref(false) const formatSelectorFilePath = ref('') async function autoGenerateSessionIndex(sessionId: string, importMode?: ImportResult['importMode']) { - if (importMode === 'incremental') return + if (importMode !== 'created') return try { const gapThreshold = getSessionGapThreshold() await useSessionIndexService().generate(sessionId, gapThreshold) @@ -63,11 +63,20 @@ async function autoGenerateSessionIndex(sessionId: string, importMode?: ImportRe } function showImportOutcome(result: ImportResult) { - if (result.importMode !== 'incremental') { + if (result.importMode === 'created') { + if (result.createReason === 'ambiguous') { + toast.warn(t('home.import.outcome.ambiguous')) + return + } toast.success(t('home.import.outcome.created')) return } + if (result.importMode !== 'incremental') { + toast.success(t('home.import.batch.success')) + return + } + const newMessageCount = result.newMessageCount ?? 0 const duplicateCount = result.duplicateCount ?? 0 if (newMessageCount === 0) { @@ -591,6 +600,7 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { file.error = result.error file.importMode = result.importMode file.matchedBy = result.matchedBy + file.createReason = result.createReason file.newMessageCount = result.newMessageCount file.duplicateCount = result.duplicateCount }, @@ -694,6 +704,7 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { batchFiles.value[i].sessionId = result.sessionId batchFiles.value[i].importMode = result.importMode batchFiles.value[i].matchedBy = result.matchedBy + batchFiles.value[i].createReason = result.createReason batchFiles.value[i].newMessageCount = result.newMessageCount batchFiles.value[i].duplicateCount = result.duplicateCount successCount++ @@ -726,6 +737,7 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { error: f.error, importMode: f.importMode, matchedBy: f.matchedBy, + createReason: f.createReason, newMessageCount: f.newMessageCount, duplicateCount: f.duplicateCount, })), @@ -845,7 +857,10 @@ function getBatchFileProgressText(file: BatchFileInfo): string { } function getBatchFileOutcomeText(file: BatchFileInfo): string { - if (file.importMode !== 'incremental') return t('home.import.batch.created') + if (file.importMode === 'created') { + return file.createReason === 'ambiguous' ? t('home.import.batch.ambiguous') : t('home.import.batch.created') + } + if (file.importMode !== 'incremental') return t('home.import.batch.success') if ((file.newMessageCount ?? 0) === 0) return t('home.import.batch.unchanged') return t('home.import.batch.incremental', { newCount: (file.newMessageCount ?? 0).toLocaleString(), diff --git a/src/pages/home/components/import/preparedImportFlow.test.ts b/src/pages/home/components/import/preparedImportFlow.test.ts index 743cb0fd..8623f9c2 100644 --- a/src/pages/home/components/import/preparedImportFlow.test.ts +++ b/src/pages/home/components/import/preparedImportFlow.test.ts @@ -25,6 +25,7 @@ describe('prepared import batch flow', () => { sessionId: `session-${chatId}`, importMode: 'incremental' as const, matchedBy: 'stable-id' as const, + createReason: undefined, newMessageCount: 2, duplicateCount: 5, } @@ -43,12 +44,14 @@ describe('prepared import batch flow', () => { { importMode: result.items[0].importMode, matchedBy: result.items[0].matchedBy, + createReason: result.items[0].createReason, newMessageCount: result.items[0].newMessageCount, duplicateCount: result.items[0].duplicateCount, }, { importMode: 'incremental', matchedBy: 'stable-id', + createReason: undefined, newMessageCount: 2, duplicateCount: 5, } diff --git a/src/pages/home/components/import/preparedImportFlow.ts b/src/pages/home/components/import/preparedImportFlow.ts index 69bb2e68..f0280dfc 100644 --- a/src/pages/home/components/import/preparedImportFlow.ts +++ b/src/pages/home/components/import/preparedImportFlow.ts @@ -17,6 +17,7 @@ export interface PreparedBatchImportResult { error?: string importMode?: ImportResult['importMode'] matchedBy?: ImportResult['matchedBy'] + createReason?: ImportResult['createReason'] newMessageCount?: number duplicateCount?: number }> @@ -24,7 +25,7 @@ export interface PreparedBatchImportResult { type PreparedChatImportResult = Pick< ImportResult, - 'success' | 'sessionId' | 'error' | 'importMode' | 'matchedBy' | 'newMessageCount' | 'duplicateCount' + 'success' | 'sessionId' | 'error' | 'importMode' | 'matchedBy' | 'createReason' | 'newMessageCount' | 'duplicateCount' > export async function runPreparedImportBatch(options: { @@ -59,6 +60,7 @@ export async function runPreparedImportBatch(options: { error: result.error, importMode: result.importMode, matchedBy: result.matchedBy, + createReason: result.createReason, newMessageCount: result.newMessageCount, duplicateCount: result.duplicateCount, }) diff --git a/src/services/import/electron.ts b/src/services/import/electron.ts index 86752e40..9b05c42b 100644 --- a/src/services/import/electron.ts +++ b/src/services/import/electron.ts @@ -60,6 +60,7 @@ export class ElectronImportAdapter implements ImportAdapter { error: result.error, importMode: result.importMode, matchedBy: result.matchedBy, + createReason: result.createReason, newMessageCount: result.newMessageCount, duplicateCount: result.duplicateCount, diagnostics: result.diagnostics, @@ -198,6 +199,7 @@ export class ElectronImportAdapter implements ImportAdapter { error: result.error, importMode: result.importMode, matchedBy: result.matchedBy, + createReason: result.createReason, newMessageCount: result.newMessageCount, duplicateCount: result.duplicateCount, diagnostics: result.diagnostics, diff --git a/src/services/import/types.ts b/src/services/import/types.ts index e9d3e547..be15cc31 100644 --- a/src/services/import/types.ts +++ b/src/services/import/types.ts @@ -18,6 +18,7 @@ export interface ImportOptions { export type AutoImportMode = 'created' | 'incremental' export type AutoImportMatchMethod = 'source-session-id' | 'stable-id' | 'trailing-messages' +export type AutoImportCreateReason = 'no-match' | 'ambiguous' export interface ImportResult { success: boolean @@ -25,6 +26,7 @@ export interface ImportResult { error?: string importMode?: AutoImportMode matchedBy?: AutoImportMatchMethod + createReason?: AutoImportCreateReason newMessageCount?: number duplicateCount?: number messageCount?: number @@ -39,6 +41,7 @@ export function normalizeImportResult(result: ImportResult): ImportResult { error: result.error, importMode: result.importMode, matchedBy: result.matchedBy, + createReason: result.createReason, newMessageCount: result.newMessageCount, duplicateCount: result.duplicateCount, messageCount: result.messageCount, @@ -52,6 +55,7 @@ export interface ImportDiagnosticsInfo { detectedFormat: string | null messagesReceived: number messagesWritten: number + duplicateCount: number messagesSkipped: number skipReasons: { noSenderId: number diff --git a/src/services/index.ts b/src/services/index.ts index 579d95f6..295855dc 100644 --- a/src/services/index.ts +++ b/src/services/index.ts @@ -25,6 +25,7 @@ export type { ImportResult, AutoImportMode, AutoImportMatchMethod, + AutoImportCreateReason, ImportDiagnosticsInfo, FormatInfo, MultiChatEntry, diff --git a/src/stores/session.ts b/src/stores/session.ts index 6a1613d5..ade8d197 100644 --- a/src/stores/session.ts +++ b/src/stores/session.ts @@ -2,7 +2,7 @@ import { defineStore } from 'pinia' import { ref, computed } from 'vue' import type { AnalysisSession, ImportProgress, ChatType } from '@/types/base' import { useDataService, useImportService, useSessionIndexService, usePlatformService } from '@/services' -import type { AutoImportMatchMethod, AutoImportMode } from '@/services' +import type { AutoImportCreateReason, AutoImportMatchMethod, AutoImportMode } from '@/services' import { IS_ELECTRON } from '@/utils/platform' /** 侧边栏筛选类型 */ @@ -44,6 +44,7 @@ export interface BatchFileInfo { sessionId?: string importMode?: AutoImportMode matchedBy?: AutoImportMatchMethod + createReason?: AutoImportCreateReason newMessageCount?: number duplicateCount?: number } @@ -191,6 +192,7 @@ export const useSessionStore = defineStore( error?: string importMode?: AutoImportMode matchedBy?: AutoImportMatchMethod + createReason?: AutoImportCreateReason newMessageCount?: number duplicateCount?: number }> { @@ -217,6 +219,7 @@ export const useSessionStore = defineStore( detectedFormat: string | null messagesReceived: number messagesWritten: number + duplicateCount: number messagesSkipped: number skipReasons: { noSenderId: number @@ -235,6 +238,7 @@ export const useSessionStore = defineStore( diagnostics?: ImportDiagnosticsInfo importMode?: AutoImportMode matchedBy?: AutoImportMatchMethod + createReason?: AutoImportCreateReason newMessageCount?: number duplicateCount?: number }> { @@ -312,6 +316,7 @@ export const useSessionStore = defineStore( diagnostics: importResult.diagnostics, importMode: importResult.importMode, matchedBy: importResult.matchedBy, + createReason: importResult.createReason, newMessageCount: importResult.newMessageCount, duplicateCount: importResult.duplicateCount, } @@ -438,6 +443,7 @@ export const useSessionStore = defineStore( file.sessionId = importResult.sessionId file.importMode = importResult.importMode file.matchedBy = importResult.matchedBy + file.createReason = importResult.createReason file.newMessageCount = importResult.newMessageCount file.duplicateCount = importResult.duplicateCount successCount++ @@ -466,6 +472,7 @@ export const useSessionStore = defineStore( file.sessionId = importResult.sessionId file.importMode = importResult.importMode file.matchedBy = importResult.matchedBy + file.createReason = importResult.createReason file.newMessageCount = importResult.newMessageCount file.duplicateCount = importResult.duplicateCount successCount++