refactor(import): simplify demo import implementation

apps/desktop/main/worker/workerManager.ts

@@ -6,21 +6,17 @@
 import { Worker } from 'worker_threads'
 import { app } from 'electron'
 import * as path from 'path'
-import * as fs from 'fs'
 import type { ParseProgress } from '@openchatlab/parser'
 import type { AutoImportResult, StreamImportResult } from './import'
 
 import { getDatabaseDir, getCacheDir, getTempDir, getLogsDir, ensureDir } from '../paths/locations'
 import { resolveDesktopNativeBinding } from '../runtime/native-sqlite'
-import { assertDesktopDataDirCompatible, getDesktopAppVersion } from '../runtime/compat'
+import { assertDesktopDataDirCompatible, createDesktopRuntimeIdentity, getDesktopAppVersion } from '../runtime/compat'
 import { getPathProvider } from '../paths/provider'
 import {
+  DatabaseManager,
   IMPORT_IN_PROGRESS_ERROR_KEY,
   ImportInProgressError,
-  contactsService,
-  deleteSessionCache,
-  globalInsightService,
-  peopleRelationshipsService,
   raiseChatDbCompatibilityGate,
   withDataDirImportLock,
 } from '@openchatlab/node-runtime'
@@ -666,24 +662,8 @@ export async function analyzePushImport(
 }
 
 export function deleteImportedSession(sessionId: string): void {
-  const dbPath = path.join(getDbDir(), `${sessionId}.db`)
-  for (const suffix of ['', '-wal', '-shm']) {
-    try {
-      const p = dbPath + suffix
-      if (fs.existsSync(p)) fs.unlinkSync(p)
-    } catch {
-      /* ignore */
-    }
-  }
-
-  const cacheDir = getCacheDir()
-  const userDataDir = getPathProvider().getUserDataDir()
-  deleteSessionCache(sessionId, cacheDir)
-  deleteSessionCache(sessionId, path.join(cacheDir, 'query'))
-  deleteSessionCache(sessionId, contactsService.getContactsFactsCacheDir(userDataDir))
-  deleteSessionCache(sessionId, globalInsightService.getGlobalInsightFactsCacheDir(userDataDir))
-  globalInsightService.deleteAnnualSummarySnapshots(globalInsightService.getGlobalInsightDir(userDataDir))
-  deleteSessionCache(sessionId, peopleRelationshipsService.getPeopleRelationshipsFactsCacheDir(userDataDir))
+  const runtime = createDesktopRuntimeIdentity(getDesktopAppVersion(app.getVersion()))
+  new DatabaseManager(getPathProvider(), { runtime }).deleteSessionDatabaseFiles(sessionId)
 }
 
 /**

packages/node-runtime/src/import/streaming-importer.test.ts

@@ -8,7 +8,7 @@ import { CHAT_DB_TABLES } from '@openchatlab/core'
 import { BetterSqliteAdapter } from '../better-sqlite3-adapter'
 import { computeAndSetOverviewCache } from '../cache/session-cache'
 import { TEMP_DB_SCHEMA } from '../merger/temp-db'
-import { analyzeNewImport, streamingImport, streamParseFileInfo } from './streaming-importer'
+import { analyzeNewImport, streamingImport, streamParseFileInfo, type StreamImportDeps } from './streaming-importer'
 
 const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
 
@@ -17,6 +17,28 @@ function makeTempDir(): string {
   return fs.mkdtempSync(path.join(baseDir, 'chatlab-streaming-import-'))
 }
 
+function createImportDeps(dbPath: string): StreamImportDeps {
+  return {
+    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 focused importer tests */
+    },
+  }
+}
+
 function writeChunkedQqExport(root: string): string {
   const chunksDir = path.join(root, 'chunks')
   fs.mkdirSync(chunksDir, { recursive: true })
@@ -223,30 +245,7 @@ test('streamingImport updates avatars for members first created from message bat
   const manifestPath = writeChunkedQqExport(root)
   const dbPath = path.join(root, 'avatar-test.db')
 
-  const result = await streamingImport(
-    manifestPath,
-    {
-      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,
-    'avatar-test'
-  )
+  const result = await streamingImport(manifestPath, createImportDeps(dbPath), undefined, 'avatar-test')
 
   assert.equal(result.success, true)
 
@@ -271,30 +270,7 @@ test('streamingImport applies incremental-equivalent deduplication on first impo
   assert.equal(analysis.newMessageCount, 3)
   assert.equal(analysis.duplicateCount, 2)
 
-  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'
-  )
+  const result = await streamingImport(filePath, createImportDeps(dbPath), undefined, 'duplicate-test')
 
   assert.equal(result.success, true)
   assert.equal(result.diagnostics?.messagesReceived, 5)
@@ -313,30 +289,7 @@ test('streamingImport canonicalizes reserved SYSTEM senders and excludes them fr
   const filePath = writeSystemChatLabExport(root)
   const dbPath = path.join(root, 'system-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,
-    'system-test'
-  )
+  const result = await streamingImport(filePath, createImportDeps(dbPath), undefined, 'system-test')
 
   assert.equal(result.success, true)
 
@@ -368,30 +321,7 @@ test('streamingImport preserves an ordinary WhatsApp participant named SYSTEM',
   const filePath = writeWhatsAppSystemParticipantExport(root)
   const dbPath = path.join(root, 'whatsapp-system-participant.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,
-    'whatsapp-system-participant'
-  )
+  const result = await streamingImport(filePath, createImportDeps(dbPath), undefined, 'whatsapp-system-participant')
 
   assert.equal(result.success, true)
 

packages/parser/src/chatlab-demo-timeline.test.ts

@@ -51,9 +51,8 @@ describe('rebaseChatLabDemoDocuments', () => {
     expectedLatest.setDate(expectedLatest.getDate() - 1)
     expectedLatest.setHours(22, 30, 0, 0)
 
-    assert.equal(result.sourceLatestTimestamp, groupTimestamps[1])
     assert.equal(result.latestTimestamp, Math.floor(expectedLatest.getTime() / 1000))
-    assert.equal(result.offsetSeconds, result.latestTimestamp - result.sourceLatestTimestamp)
+    assert.equal(result.offsetSeconds, result.latestTimestamp - groupTimestamps[1])
     assert.deepEqual(
       rebased[0].messages.map((message: { timestamp: number }) => message.timestamp),
       groupTimestamps.map((timestamp) => timestamp + result.offsetSeconds)

packages/parser/src/chatlab-demo-timeline.ts

@@ -23,7 +23,6 @@ interface ChatLabDemoDocument extends Record<string, unknown> {
 export interface RebasedChatLabDemoDocuments {
   documents: string[]
   offsetSeconds: number
-  sourceLatestTimestamp: number
   latestTimestamp: number
 }
 
@@ -64,7 +63,6 @@ export function rebaseChatLabDemoDocuments(
   return {
     documents: documents.map((document) => JSON.stringify(document)),
     offsetSeconds,
-    sourceLatestTimestamp,
     latestTimestamp,
   }
 }

src/services/import/browser.test.ts

@@ -272,7 +272,6 @@ describe('BrowserImportAdapter', () => {
 
   it('deletes sessions created earlier in the batch when a later demo import fails', async () => {
     const deletedSessionIds: string[] = []
-    const reportedErrors: unknown[] = []
     const importError = new Error('third import failed')
     let importCount = 0
     const documents = [
@@ -309,13 +308,7 @@ describe('BrowserImportAdapter', () => {
     const adapter = new BrowserImportAdapter(
       rpc,
       async () => new Response(documents[downloadIndex++], { status: 200 }),
-      () => new Date('2026-07-25T04:00:00.000Z'),
-      60_000,
-      (event) => {
-        if (event.level === 'error' && event.message === 'Demo import failed') {
-          reportedErrors.push(event.data?.error)
-        }
-      }
+      () => new Date('2026-07-25T04:00:00.000Z')
     )
 
     const result = await adapter.importDemo('en')
@@ -323,7 +316,6 @@ describe('BrowserImportAdapter', () => {
     assert.equal(result.success, false)
     assert.match(result.error ?? '', /third import failed/)
     assert.deepEqual(deletedSessionIds, ['session-2', 'session-1'])
-    assert.deepEqual(reportedErrors, [importError])
   })
 
   it('forwards browser-safe format identifiers to the worker runtime', async () => {

src/services/import/browser.ts

@@ -32,8 +32,7 @@ export class BrowserImportAdapter implements ImportAdapter {
     private readonly rpc: BrowserRuntimeRpcPort,
     private readonly fetchDemo: typeof fetch = fetch,
     private readonly now: () => Date = () => new Date(),
-    private readonly demoDownloadTimeoutMs = DEMO_DOWNLOAD_TIMEOUT_MS,
-    private readonly reportLog: typeof reportRuntimeLog = reportRuntimeLog
+    private readonly demoDownloadTimeoutMs = DEMO_DOWNLOAD_TIMEOUT_MS
   ) {}
 
   async importFile(
@@ -101,27 +100,20 @@ export class BrowserImportAdapter implements ImportAdapter {
 
     const controller = new AbortController()
     this.activeImport = controller
-    this.reportLog({ level: 'info', scope: 'web-wasm-demo', message: 'Demo import started' })
+    reportRuntimeLog({ level: 'info', scope: 'web-wasm-demo', message: 'Demo import started' })
     const sessionIds: string[] = []
 
     try {
-      const downloaded: Array<{ filename: string; contentType: string; json: string }> = []
+      const downloaded: string[] = []
       for (const filename of DEMO_FILES) {
         onProgress?.({ stage: 'downloading' })
-        const { content, contentType } = await this.downloadDemoFile(locale, filename, controller)
-        downloaded.push({
-          filename,
-          contentType,
-          json: new TextDecoder().decode(content),
-        })
+        const content = await this.downloadDemoFile(locale, filename, controller)
+        downloaded.push(new TextDecoder().decode(content))
       }
 
-      const rebased = rebaseChatLabDemoDocuments(
-        downloaded.map(({ json }) => json),
-        this.now()
-      )
+      const rebased = rebaseChatLabDemoDocuments(downloaded, this.now())
       const files = rebased.documents.map(
-        (json, index) => new File([json], downloaded[index].filename, { type: downloaded[index].contentType })
+        (json, index) => new File([json], DEMO_FILES[index], { type: 'application/json' })
       )
 
       for (const file of files) {
@@ -134,7 +126,7 @@ export class BrowserImportAdapter implements ImportAdapter {
       }
 
       const [groupSessionId, ...privateSessionIds] = sessionIds
-      this.reportLog({
+      reportRuntimeLog({
         level: 'info',
         scope: 'web-wasm-demo',
         message: 'Demo import completed',
@@ -148,7 +140,7 @@ export class BrowserImportAdapter implements ImportAdapter {
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error)
       const rollbackFailureCount = await this.rollbackDemoSessions(sessionIds)
-      this.reportLog({
+      reportRuntimeLog({
         level: 'error',
         scope: 'web-wasm-demo',
         message: 'Demo import failed',
@@ -170,7 +162,7 @@ export class BrowserImportAdapter implements ImportAdapter {
     locale: string,
     filename: (typeof DEMO_FILES)[number],
     controller: AbortController
-  ): Promise<{ content: ArrayBuffer; contentType: string }> {
+  ): Promise<ArrayBuffer> {
     // Use one controller for user cancellation and per-file timeout so response-body reads are also abortable.
     const timeout = setTimeout(() => {
       controller.abort(new Error(`Download demo timed out (${filename})`))
@@ -184,10 +176,7 @@ export class BrowserImportAdapter implements ImportAdapter {
 
       const content = await response.arrayBuffer()
       if (content.byteLength < 100) throw new Error(`Downloaded demo file is too small (${filename})`)
-      return {
-        content,
-        contentType: response.headers.get('content-type') || 'application/json',
-      }
+      return content
     } finally {
       clearTimeout(timeout)
     }
@@ -250,7 +239,7 @@ export class BrowserImportAdapter implements ImportAdapter {
     }
 
     if (sessionIds.length > 0) {
-      this.reportLog({
+      reportRuntimeLog({
         level: failureCount > 0 ? 'error' : 'info',
         scope: 'web-wasm-demo',
         message: failureCount > 0 ? 'Demo import rollback incomplete' : 'Demo import rolled back',