diff --git a/apps/desktop/main/api/routes/import-helpers.test.ts b/apps/desktop/main/api/routes/import-helpers.test.ts index 961d4444..072863c5 100644 --- a/apps/desktop/main/api/routes/import-helpers.test.ts +++ b/apps/desktop/main/api/routes/import-helpers.test.ts @@ -2,7 +2,7 @@ import assert from 'node:assert/strict' import test from 'node:test' import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime/src/import/import-lock' import { ApiErrorCode } from '../errors' -import { apiErrorFromImportResult, batchFromStreamDiagnostics } from './import-helpers' +import { analysisFromNewImport, apiErrorFromImportResult, batchFromStreamDiagnostics } from './import-helpers' test('maps an import lock result to the documented 409 API error', () => { const error = apiErrorFromImportResult(IMPORT_IN_PROGRESS_ERROR_KEY, 'Import failed') @@ -42,3 +42,21 @@ test('maps stream duplicate and invalid-row counts to their distinct API fields' } ) }) + +test('maps new-import analysis counts without assuming every parsed message is new', () => { + assert.deepEqual( + analysisFromNewImport({ + totalMessages: 12, + newMessageCount: 7, + duplicateCount: 3, + totalMembers: 4, + meta: { name: 'Fixture', platform: 'wechat', type: 'private' }, + }), + { + totalInFile: 12, + newMessageCount: 7, + duplicateCount: 3, + newMemberCount: 4, + } + ) +}) diff --git a/apps/desktop/main/api/routes/import-helpers.ts b/apps/desktop/main/api/routes/import-helpers.ts index 096225f3..7030d08f 100644 --- a/apps/desktop/main/api/routes/import-helpers.ts +++ b/apps/desktop/main/api/routes/import-helpers.ts @@ -1,5 +1,5 @@ import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime/src/import/import-lock' -import type { ImportDiagnostics } from '@openchatlab/node-runtime/src/import/streaming-importer' +import type { AnalyzeNewImportResult, ImportDiagnostics } from '@openchatlab/node-runtime/src/import/streaming-importer' import { importFailed, importInProgress, type ApiError } from '../errors' export function apiErrorFromImportResult(error: string | undefined, fallbackMessage: string): ApiError { @@ -20,3 +20,17 @@ export function batchFromStreamDiagnostics(diagnostics: ImportDiagnostics | unde duplicateCount: diagnostics.duplicateCount, } } + +export function analysisFromNewImport(result: AnalyzeNewImportResult): { + totalInFile: number + newMessageCount: number + duplicateCount: number + newMemberCount: number +} { + return { + totalInFile: result.totalMessages, + newMessageCount: result.newMessageCount, + duplicateCount: result.duplicateCount, + newMemberCount: result.totalMembers, + } +} diff --git a/apps/desktop/main/api/routes/import.ts b/apps/desktop/main/api/routes/import.ts index 4cc749b3..33986026 100644 --- a/apps/desktop/main/api/routes/import.ts +++ b/apps/desktop/main/api/routes/import.ts @@ -33,7 +33,7 @@ import { errorResponse, } from '../errors' import { apiLogger } from '../logger' -import { apiErrorFromImportResult, batchFromStreamDiagnostics } from './import-helpers' +import { analysisFromNewImport, apiErrorFromImportResult, batchFromStreamDiagnostics } from './import-helpers' // Per-session lock: different sessionIds can import in parallel. const isImporting = new Set() @@ -252,12 +252,7 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply, sessionId, created: true, dryRun: true, - analysis: { - totalInFile: result.totalMessages, - newMessageCount: result.totalMessages, - duplicateCount: 0, - newMemberCount: result.totalMembers, - }, + analysis: analysisFromNewImport(result), }) } idempotencySuccess(cacheKey, responsePayload) diff --git a/apps/desktop/main/worker/import/streamImport.ts b/apps/desktop/main/worker/import/streamImport.ts index 7225a70a..25ebbba9 100644 --- a/apps/desktop/main/worker/import/streamImport.ts +++ b/apps/desktop/main/worker/import/streamImport.ts @@ -161,6 +161,8 @@ export async function analyzeNewImport( requestId: string ): Promise<{ totalMessages: number + newMessageCount: number + duplicateCount: number totalMembers: number meta: { name: string; platform: string; type: string } | null error?: string diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index b03bba07..ea7413ff 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/apps/desktop/main/worker/workerManager.ts @@ -1031,6 +1031,8 @@ export async function incrementalImport( */ export interface AnalyzeNewImportResult { totalMessages: number + newMessageCount: number + duplicateCount: number totalMembers: number meta: { name: string; platform: string; type: string } | null error?: string diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index bb92fe73..b73cfdc7 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/packages/node-runtime/src/import/streaming-importer.test.ts @@ -6,7 +6,7 @@ import test from 'node:test' import Database from 'better-sqlite3' import { CHAT_DB_TABLES } from '@openchatlab/core' import { BetterSqliteAdapter } from '../better-sqlite3-adapter' -import { streamingImport } from './streaming-importer' +import { analyzeNewImport, streamingImport } from './streaming-importer' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -175,6 +175,11 @@ test('streamingImport applies incremental-equivalent deduplication on first impo const filePath = writeDuplicateChatLabExport(root) const dbPath = path.join(root, 'duplicate-test.db') + const analysis = await analyzeNewImport(filePath, () => {}) + assert.equal(analysis.totalMessages, 5) + assert.equal(analysis.newMessageCount, 3) + assert.equal(analysis.duplicateCount, 2) + const result = await streamingImport( filePath, { diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index 74432d83..715ab235 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -25,7 +25,7 @@ import { } from '@openchatlab/parser' import * as fs from 'fs' import { buildFtsIndex } from '../fts' -import { createMessageDedupState, registerMessageAndCheckDuplicate } from './message-deduplicator' +import { createMessageDedupState, registerMessageAndCheckDuplicate, type DedupMessage } from './message-deduplicator' // ==================== Public interfaces ==================== @@ -99,6 +99,39 @@ function defaultGenerateSessionId(): string { return `chat_${ts}_${rand}` } +type CreateMessageSkipCounter = + | 'skippedNoSenderId' + | 'skippedNoAccountName' + | 'skippedInvalidTimestamp' + | 'skippedNoType' + +type PreparedCreateMessage = { message: DedupMessage } | { skipCounter: CreateMessageSkipCounter } + +function prepareMessageForCreate(message: ParsedMessage): PreparedCreateMessage { + if (!message.senderPlatformId) return { skipCounter: 'skippedNoSenderId' } + if (!message.senderAccountName) return { skipCounter: 'skippedNoAccountName' } + if (message.timestamp === undefined || message.timestamp === null || isNaN(message.timestamp)) { + return { skipCounter: 'skippedInvalidTimestamp' } + } + if (message.type === undefined || message.type === null) return { skipCounter: 'skippedNoType' } + + let content: string | null = null + if (message.content != null) { + content = typeof message.content === 'string' ? message.content : JSON.stringify(message.content) + } + + return { + message: { + platformMessageId: message.platformMessageId, + timestamp: message.timestamp, + senderPlatformId: message.senderPlatformId, + type: message.type, + content, + replyToMessageId: message.replyToMessageId, + }, + } +} + /** * High-performance streaming import: parse a file and write to DB * with batched transactions. Supports format auto-detection with fallback. @@ -384,41 +417,15 @@ async function streamImportSingle( let nicknameChangeCount = 0 for (const msg of messages) { - if (!msg.senderPlatformId) { - callbackStats.skippedNoSenderId++ - continue - } - if (!msg.senderAccountName) { - callbackStats.skippedNoAccountName++ - continue - } - if (msg.timestamp === undefined || msg.timestamp === null || isNaN(msg.timestamp)) { - callbackStats.skippedInvalidTimestamp++ - continue - } - if (msg.type === undefined || msg.type === null) { - callbackStats.skippedNoType++ + const prepared = prepareMessageForCreate(msg) + if ('skipCounter' in prepared) { + callbackStats[prepared.skipCounter]++ continue } - let safeContent: string | null = null - if (msg.content != null) { - safeContent = typeof msg.content === 'string' ? msg.content : JSON.stringify(msg.content) - } + const dedupMessage = prepared.message - if ( - registerMessageAndCheckDuplicate( - { - platformMessageId: msg.platformMessageId, - timestamp: msg.timestamp, - senderPlatformId: msg.senderPlatformId, - type: msg.type, - content: safeContent, - replyToMessageId: msg.replyToMessageId, - }, - dedupState - ) - ) { + if (registerMessageAndCheckDuplicate(dedupMessage, dedupState)) { duplicateCount++ continue } @@ -450,9 +457,9 @@ async function streamImportSingle( senderId, msg.senderAccountName || null, msg.senderGroupNickname || null, - msg.timestamp, - msg.type, - safeContent, + dedupMessage.timestamp, + dedupMessage.type, + dedupMessage.content, msg.replyToMessageId || null, msg.platformMessageId || null ) @@ -665,6 +672,8 @@ async function streamImportSingle( export interface AnalyzeNewImportResult { totalMessages: number + newMessageCount: number + duplicateCount: number totalMembers: number meta: { name: string; platform: string; type: string } | null error?: string @@ -676,12 +685,22 @@ export async function analyzeNewImport( ): Promise { const formatFeature = detectFormat(filePath) if (!formatFeature) { - return { totalMessages: 0, totalMembers: 0, meta: null, error: 'error.unrecognized_format' } + return { + totalMessages: 0, + newMessageCount: 0, + duplicateCount: 0, + totalMembers: 0, + meta: null, + error: 'error.unrecognized_format', + } } let meta: { name: string; platform: string; type: string } | null = null const memberSet = new Set() + const dedupState = createMessageDedupState() let totalMessages = 0 + let newMessageCount = 0 + let duplicateCount = 0 await streamParseFile(filePath, { onMeta: (parsedMeta: ParsedMeta) => { @@ -697,11 +716,19 @@ export async function analyzeNewImport( for (const msg of batch) { totalMessages++ if (!memberSet.has(msg.senderPlatformId)) memberSet.add(msg.senderPlatformId) + + const prepared = prepareMessageForCreate(msg) + if ('skipCounter' in prepared) continue + if (registerMessageAndCheckDuplicate(prepared.message, dedupState)) { + duplicateCount++ + } else { + newMessageCount++ + } } }, }) - return { totalMessages, totalMembers: memberSet.size, meta } + return { totalMessages, newMessageCount, duplicateCount, totalMembers: memberSet.size, meta } } // ==================== Temp DB for merge preview ====================