fix(api): align new import dry runs

Reuse create-path validation and deduplication during new-session analysis so dry-run counts match the subsequent import.

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,
+    }
+  )
+})

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,
+  }
+}

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<string>()
@@ -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)

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

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

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,
     {

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<AnalyzeNewImportResult> {
   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<string>()
+  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 ====================