fix(import): wire batch adapters and failure diagnostics

Preserve batch support through telemetry, forward item-scoped progress across Desktop and CLI Web, and retain original backend failure details for diagnostics and unfinished results.

src/services/import/fetch.ts

@@ -142,6 +142,7 @@ export class FetchImportAdapter implements ImportAdapter {
       const decoder = new TextDecoder()
       let buffer = ''
       let results: BatchImportItemResult[] | null = null
+      let terminalError: string | null = null
       let eventType = ''
 
       while (true) {
@@ -167,6 +168,8 @@ export class FetchImportAdapter implements ImportAdapter {
             onProgress?.({ index: data.index, event: 'complete', result })
           } else if (eventType === 'done') {
             results = (data as unknown[]).map((result, index) => normalizeBatchResult(items, index, result))
+          } else if (eventType === 'error') {
+            terminalError = typeof data?.error === 'string' ? data.error : 'Batch import failed'
           }
           eventType = ''
         }
@@ -178,7 +181,7 @@ export class FetchImportAdapter implements ImportAdapter {
             completedResults[index] ?? {
               id: item.id,
               status: 'failed',
-              error: 'Batch response ended without results',
+              error: terminalError ?? 'Batch response ended without results',
             }
         )
       )

src/services/import/importSourceAdapters.test.ts

@@ -270,6 +270,49 @@ describe('archive import source adapters', () => {
     assert.equal(progress.length, 2)
   })
 
+  it('preserves a server batch error for items without completed results', async () => {
+    globalThis.fetch = (async () =>
+      new Response(
+        [
+          'event: batch-complete',
+          'data: {"index":0,"result":{"id":"0","status":"success","result":{"success":true,"sessionId":"session-a"}}}',
+          '',
+          'event: error',
+          'data: {"error":"compatibility gate unavailable"}',
+          '',
+          '',
+        ].join('\n'),
+        { headers: { 'Content-Type': 'text/event-stream' } }
+      )) as typeof fetch
+
+    const results = await new FetchImportAdapter().importBatch?.([
+      { id: 'first', file: new File(['{}'], 'first.json') },
+      { id: 'second', file: new File(['{}'], 'second.json') },
+    ])
+
+    assert.deepEqual(results, [
+      {
+        id: 'first',
+        status: 'success',
+        result: {
+          success: true,
+          sessionId: 'session-a',
+          platform: undefined,
+          error: undefined,
+          importMode: undefined,
+          matchedBy: undefined,
+          createReason: undefined,
+          newMessageCount: undefined,
+          duplicateCount: undefined,
+          messageCount: undefined,
+          memberCount: undefined,
+          diagnostics: undefined,
+        },
+      },
+      { id: 'second', status: 'failed', error: 'compatibility gate unavailable' },
+    ])
+  })
+
   it('keeps the Web batch stream open until cancellation returns authoritative results', async () => {
     const requests: Array<{ url: string; batchId?: string }> = []
     let streamController: ReadableStreamDefaultController<Uint8Array> | undefined

src/services/import/telemetry.test.ts

@@ -28,6 +28,77 @@ function createDelegate(
 }
 
 describe('TelemetryImportAdapter', () => {
+  it('preserves and tracks the delegate batch capability', async () => {
+    const events: Array<{ name: AnalyticsEventName; properties?: Record<string, unknown> }> = []
+    const progressEvents: unknown[] = []
+    const delegate = createDelegate()
+    let cancelCalls = 0
+    let receivedOptions: unknown
+
+    delegate.importBatch = async (items, options, onProgress) => {
+      receivedOptions = options
+      onProgress?.({ index: 0, event: 'start' })
+      return [
+        {
+          id: items[0].id,
+          status: 'success',
+          result: { success: true, sessionId: 'session-1', platform: 'qq', importMode: 'created' },
+        },
+        { id: items[1].id, status: 'failed', error: 'Parser failed at /Users/alice/private.json' },
+      ]
+    }
+    delegate.cancelActiveImport = () => {
+      cancelCalls++
+    }
+
+    const platform = {
+      trackAnalyticsEvent: async (name: AnalyticsEventName, properties?: Record<string, unknown>) => {
+        events.push({ name, properties })
+      },
+    }
+    const adapter = new TelemetryImportAdapter(delegate, platform) as ImportAdapter
+    const adapterWithoutBatch = new TelemetryImportAdapter(createDelegate(), platform) as ImportAdapter
+    const items = [
+      { id: 'first', file: new File(['{}'], 'first-private.json') },
+      { id: 'second', file: new File(['{}'], 'second-private.json') },
+    ]
+
+    assert.equal(adapterWithoutBatch.importBatch, undefined)
+    assert.equal(typeof adapter.importBatch, 'function')
+    const results = await adapter.importBatch!(items, { formatId: 'whatsapp-native-txt' }, (event) =>
+      progressEvents.push(event)
+    )
+    adapter.cancelActiveImport?.()
+    await new Promise((resolve) => setTimeout(resolve, 0))
+
+    assert.deepEqual(receivedOptions, { formatId: 'whatsapp-native-txt' })
+    assert.deepEqual(progressEvents, [{ index: 0, event: 'start' }])
+    assert.deepEqual(
+      results.map((result) => ({ id: result.id, status: result.status })),
+      [
+        { id: 'first', status: 'success' },
+        { id: 'second', status: 'failed' },
+      ]
+    )
+    assert.equal(cancelCalls, 1)
+    assert.deepEqual(
+      events.map((event) => ({
+        name: event.name,
+        platform: event.properties?.chat_platform,
+        failureReason: event.properties?.failure_reason,
+      })),
+      [
+        { name: 'chat_import_started', platform: 'whatsapp', failureReason: undefined },
+        { name: 'chat_import_completed', platform: 'qq', failureReason: undefined },
+        { name: 'chat_import_started', platform: 'whatsapp', failureReason: undefined },
+        { name: 'chat_import_failed', platform: 'whatsapp', failureReason: 'parse' },
+      ]
+    )
+    assert.equal(typeof events[1].properties?.duration_ms, 'number')
+    assert.equal(JSON.stringify(events).includes('private.json'), false)
+    assert.equal(JSON.stringify(events).includes('/Users/alice'), false)
+  })
+
   it('tracks the remembered format and the actual imported platform without file details', async () => {
     const events: Array<{ name: AnalyticsEventName; properties?: Record<string, unknown> }> = []
     const file = new File(['{}'], 'private-chat.json')

src/services/import/telemetry.ts

@@ -1,6 +1,9 @@
 import type { AnalyticsEventName } from '@openchatlab/shared-types'
 import type { PlatformAdapter } from '../platform/types'
 import type {
+  BatchImportItem,
+  BatchImportItemResult,
+  BatchImportProgress,
   DemoImportResult,
   DemoProgress,
   FormatInfo,
@@ -57,11 +60,18 @@ export class TelemetryImportAdapter implements ImportAdapter {
   private readonly filePlatforms = new WeakMap<File, string>()
   private readonly pathPlatforms = new Map<string, string>()
   private readonly sourcePlatforms = new Map<string, string>()
+  readonly importBatch?: NonNullable<ImportAdapter['importBatch']>
 
   constructor(
     private readonly delegate: ImportAdapter,
     private readonly platform: Pick<PlatformAdapter, 'trackAnalyticsEvent'>
-  ) {}
+  ) {
+    const importBatch = delegate.importBatch?.bind(delegate)
+    if (importBatch) {
+      this.importBatch = (items, options, onProgress) =>
+        this.runTrackedBatchImport(importBatch, items, options, onProgress)
+    }
+  }
 
   private track(eventName: AnalyticsEventName, properties?: Record<string, unknown>): void {
     void this.platform.trackAnalyticsEvent(eventName, properties).catch(() => {})
@@ -80,37 +90,80 @@ export class TelemetryImportAdapter implements ImportAdapter {
     else this.filePlatforms.set(source, platform)
   }
 
+  private trackImportResult(
+    initialPlatform: string,
+    startedAt: number,
+    result: { success: boolean; error?: string; platform?: string; importMode?: 'created' | 'incremental' }
+  ): void {
+    if (result.importMode === 'incremental') {
+      this.track('incremental_import_used')
+      return
+    }
+
+    this.track('chat_import_started', { chat_platform: initialPlatform })
+    const chatPlatform = result.platform ?? initialPlatform
+    if (result.success) {
+      this.track('chat_import_completed', {
+        chat_platform: chatPlatform,
+        duration_ms: Date.now() - startedAt,
+      })
+    } else {
+      this.track('chat_import_failed', {
+        chat_platform: chatPlatform,
+        failure_reason: classifyImportFailure(result.error),
+      })
+    }
+  }
+
+  private trackImportError(initialPlatform: string, error: unknown): void {
+    this.track('chat_import_started', { chat_platform: initialPlatform })
+    this.track('chat_import_failed', {
+      chat_platform: initialPlatform,
+      failure_reason: classifyImportFailure(error instanceof Error ? error.message : undefined),
+    })
+  }
+
   private async runTrackedImport<
     T extends { success: boolean; error?: string; platform?: string; importMode?: 'created' | 'incremental' },
   >(initialPlatform: string, operation: () => Promise<T>): Promise<T> {
     const startedAt = Date.now()
-    const trackStarted = () => this.track('chat_import_started', { chat_platform: initialPlatform })
     try {
       const result = await operation()
-      if (result.importMode === 'incremental') {
-        this.track('incremental_import_used')
-        return result
-      }
-      trackStarted()
-      const chatPlatform = result.platform ?? initialPlatform
-      if (result.success) {
-        this.track('chat_import_completed', {
-          chat_platform: chatPlatform,
-          duration_ms: Date.now() - startedAt,
-        })
-      } else {
-        this.track('chat_import_failed', {
-          chat_platform: chatPlatform,
-          failure_reason: classifyImportFailure(result.error),
-        })
-      }
+      this.trackImportResult(initialPlatform, startedAt, result)
       return result
     } catch (error) {
-      trackStarted()
-      this.track('chat_import_failed', {
-        chat_platform: initialPlatform,
-        failure_reason: classifyImportFailure(error instanceof Error ? error.message : undefined),
-      })
+      this.trackImportError(initialPlatform, error)
+      throw error
+    }
+  }
+
+  private async runTrackedBatchImport(
+    importBatch: NonNullable<ImportAdapter['importBatch']>,
+    items: BatchImportItem[],
+    options?: ImportOptions,
+    onProgress?: (progress: BatchImportProgress) => void
+  ): Promise<BatchImportItemResult[]> {
+    const startedAt = Date.now()
+    const itemsById = new Map(items.map((item) => [item.id, item]))
+    try {
+      const results = await importBatch(items, options, onProgress)
+      for (const result of results) {
+        if (result.status === 'cancelled') continue
+        const item = itemsById.get(result.id)
+        const initialPlatform = item
+          ? this.rememberedPlatform(item.file, options)
+          : platformFromFormatId(options?.formatId)
+        if (result.status === 'success') {
+          this.trackImportResult(initialPlatform, startedAt, result.result)
+        } else if (result.result) {
+          this.trackImportResult(initialPlatform, startedAt, result.result)
+        } else {
+          this.trackImportResult(initialPlatform, startedAt, { success: false, error: result.error })
+        }
+      }
+      return results
+    } catch (error) {
+      for (const item of items) this.trackImportError(this.rememberedPlatform(item.file, options), error)
       throw error
     }
   }