From 1f0528ebd05719b446f0ab4dd72e4b8fa3cdea94 Mon Sep 17 00:00:00 2001 From: digua Date: Sun, 26 Jul 2026 01:30:26 +0800 Subject: [PATCH] refactor(import): simplify demo import implementation --- apps/desktop/main/worker/workerManager.ts | 28 +--- .../src/import/streaming-importer.test.ts | 124 ++++-------------- .../parser/src/chatlab-demo-timeline.test.ts | 3 +- packages/parser/src/chatlab-demo-timeline.ts | 2 - src/services/import/browser.test.ts | 10 +- src/services/import/browser.ts | 35 ++--- 6 files changed, 45 insertions(+), 157 deletions(-) diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index 3b8010d8..4c316471 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/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) } /** diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index 6fddb933..1282218a 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/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) diff --git a/packages/parser/src/chatlab-demo-timeline.test.ts b/packages/parser/src/chatlab-demo-timeline.test.ts index cecc1a59..30791d48 100644 --- a/packages/parser/src/chatlab-demo-timeline.test.ts +++ b/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) diff --git a/packages/parser/src/chatlab-demo-timeline.ts b/packages/parser/src/chatlab-demo-timeline.ts index 3a969a0a..6374924e 100644 --- a/packages/parser/src/chatlab-demo-timeline.ts +++ b/packages/parser/src/chatlab-demo-timeline.ts @@ -23,7 +23,6 @@ interface ChatLabDemoDocument extends Record { 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, } } diff --git a/src/services/import/browser.test.ts b/src/services/import/browser.test.ts index 6305c8a4..571ff65c 100644 --- a/src/services/import/browser.test.ts +++ b/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 () => { diff --git a/src/services/import/browser.ts b/src/services/import/browser.ts index 4bc539df..96e86f08 100644 --- a/src/services/import/browser.ts +++ b/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 { // 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',