From 8f8a19f96ee5a74ee9e75c00c74d31210194f7f2 Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 13 Jul 2026 15:10:42 +0800 Subject: [PATCH] fix(import): align shared push import semantics --- apps/desktop/main/worker/dbWorker.ts | 2 + apps/desktop/main/worker/import/index.ts | 2 + .../main/worker/import/pushImport.test.ts | 37 ++++ apps/desktop/main/worker/import/pushImport.ts | 43 ++++ apps/desktop/main/worker/workerManager.ts | 35 +++ .../main/worker/workerTimeoutPolicy.test.ts | 1 + .../src/import/message-deduplicator.ts | 4 +- .../node-runtime/src/import/session-id.ts | 5 + .../src/services/push-importer.test.ts | 150 ++++++++++++- .../src/services/push-importer.ts | 209 ++++++++++-------- 10 files changed, 395 insertions(+), 93 deletions(-) create mode 100644 apps/desktop/main/worker/import/pushImport.test.ts create mode 100644 apps/desktop/main/worker/import/pushImport.ts create mode 100644 packages/node-runtime/src/import/session-id.ts diff --git a/apps/desktop/main/worker/dbWorker.ts b/apps/desktop/main/worker/dbWorker.ts index 730f19be..d2613a7c 100644 --- a/apps/desktop/main/worker/dbWorker.ts +++ b/apps/desktop/main/worker/dbWorker.ts @@ -81,6 +81,7 @@ import { analyzeIncrementalImport, incrementalImport, analyzeNewImport, + pushImport, } from './import' import { initNlpDir } from '@openchatlab/node-runtime' @@ -248,6 +249,7 @@ const asyncHandlers: Record Promise // 流式导入 streamImport: (p, id) => streamImport(p.filePath, id, p.formatOptions, p.externalSessionId), autoImport: (p, id) => autoImport(p.filePath, id, p.formatOptions, p.explicitSessionId), + pushImport: (p) => pushImport(p.sessionId, p.payload), // 流式解析文件信息（用于合并预览） streamParseFileInfo: (p, id) => streamParseFileInfo(p.filePath, id), // 增量导入 diff --git a/apps/desktop/main/worker/import/index.ts b/apps/desktop/main/worker/import/index.ts index f0ab1a3d..ccd3f1ad 100644 --- a/apps/desktop/main/worker/import/index.ts +++ b/apps/desktop/main/worker/import/index.ts @@ -24,6 +24,8 @@ export { type IncrementalImportResult, } from './incrementalImport' +export { pushImport } from './pushImport' + // 工具函数（供其他模块使用） export { sendProgress, generateSessionId, getDbPath, createDatabaseWithoutIndexes, createIndexes } from './utils' diff --git a/apps/desktop/main/worker/import/pushImport.test.ts b/apps/desktop/main/worker/import/pushImport.test.ts new file mode 100644 index 00000000..a50e55c9 --- /dev/null +++ b/apps/desktop/main/worker/import/pushImport.test.ts @@ -0,0 +1,37 @@ +import assert from 'node:assert/strict' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import test from 'node:test' +import { initDbDir } from '../core/dbCore' +import { pushImport } from './pushImport' + +const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') + +test('Desktop worker push import uses the shared writer and canonical database directory', async (t) => { + const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() + const root = fs.mkdtempSync(path.join(baseDir, 'chatlab-worker-push-import-')) + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + + const dbDir = path.join(root, 'databases') + initDbDir(dbDir, path.join(root, 'cache'), path.join(root, 'temp'), nativeBinding, path.join(root, 'logs')) + + const outcome = await pushImport('worker-session', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Worker Push Import', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { + platformMessageId: 'msg-1', + sender: 'wxid_alice', + timestamp: 1780330832, + type: 0, + content: 'hello', + }, + ], + }) + + assert.equal(outcome.ok, true, JSON.stringify(outcome)) + assert.equal(fs.existsSync(path.join(dbDir, 'worker-session.db')), true) + assert.equal(fs.existsSync(path.join(root, 'worker-session.db')), false) +}) diff --git a/apps/desktop/main/worker/import/pushImport.ts b/apps/desktop/main/worker/import/pushImport.ts new file mode 100644 index 00000000..4fe818e3 --- /dev/null +++ b/apps/desktop/main/worker/import/pushImport.ts @@ -0,0 +1,43 @@ +import * as fs from 'node:fs' +import * as path from 'node:path' +import { BetterSqliteAdapter } from '@openchatlab/node-runtime/src/better-sqlite3-adapter' +import { + executePushImportUnlocked, + type PushImportOutcome, + type PushImportPayload, +} from '@openchatlab/node-runtime/src/services/push-importer' +import { closeDatabase, getDbDir, openRawDatabase } from '../core/dbCore' + +function getDbPath(sessionId: string): string { + return path.join(getDbDir(), `${sessionId}.db`) +} + +export async function pushImport(sessionId: string, payload: PushImportPayload): Promise { + return executePushImportUnlocked( + { + getDbPath, + openDatabase(id, options) { + const dbPath = getDbPath(id) + if (!options.create && !fs.existsSync(dbPath)) throw new Error(`Session database not found: ${id}`) + closeDatabase(id) + fs.mkdirSync(path.dirname(dbPath), { recursive: true }) + const db = options.readonly === undefined ? openRawDatabase(dbPath) : openRawDatabase(dbPath, options) + return new BetterSqliteAdapter(db) + }, + deleteDatabase(id) { + closeDatabase(id) + const dbPath = getDbPath(id) + for (const suffix of ['', '-wal', '-shm']) { + try { + const filePath = dbPath + suffix + if (fs.existsSync(filePath)) fs.unlinkSync(filePath) + } catch { + /* best-effort cleanup */ + } + } + }, + }, + sessionId, + payload + ) +} diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index ea7413ff..14094557 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/apps/desktop/main/worker/workerManager.ts @@ -21,6 +21,7 @@ import { raiseChatDbCompatibilityGate, withDataDirImportLock, } from '@openchatlab/node-runtime' +import type { PushImportOutcome, PushImportPayload } from '@openchatlab/node-runtime' import { isRestartableReadOnlyRequestType } from './workerTimeoutPolicy' interface WorkerRequestOptions { @@ -615,6 +616,40 @@ export async function autoImport( } } +/** + * 在 worker 中执行共享 Push Import 写库语义；主进程持有全局导入锁， + * 并在 worker 成功后完成数据目录兼容门禁更新。 + */ +export async function pushImport(sessionId: string, payload: PushImportPayload): Promise { + try { + return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => { + assertDataDirCompatibleNow() + const outcome = await sendToWorker('pushImport', { sessionId, payload }) + if (!outcome.ok) return outcome + + try { + raiseChatDbCompatibilityGate(getPathProvider(), { + version: getDesktopAppVersion(app.getVersion()), + kind: 'desktop', + }) + } catch (error) { + if (outcome.result.created) deleteImportedSessionFiles(sessionId) + throw error + } + return outcome + }) + } catch (error) { + if (error instanceof ImportInProgressError) { + return { + ok: false, + reason: 'import_in_progress', + message: 'Another import is already in progress', + } + } + throw error + } +} + function deleteImportedSessionFiles(sessionId: string): void { const dbPath = path.join(getDbDir(), `${sessionId}.db`) for (const suffix of ['', '-wal', '-shm']) { diff --git a/apps/desktop/main/worker/workerTimeoutPolicy.test.ts b/apps/desktop/main/worker/workerTimeoutPolicy.test.ts index e742ff81..8194245b 100644 --- a/apps/desktop/main/worker/workerTimeoutPolicy.test.ts +++ b/apps/desktop/main/worker/workerTimeoutPolicy.test.ts @@ -21,6 +21,7 @@ describe('worker timeout policy', () => { assert.equal(isRestartableReadOnlyRequestType('deleteMember'), false) assert.equal(isRestartableReadOnlyRequestType('streamImport'), false) assert.equal(isRestartableReadOnlyRequestType('autoImport'), false) + assert.equal(isRestartableReadOnlyRequestType('pushImport'), false) assert.equal(isRestartableReadOnlyRequestType('incrementalImport'), false) assert.equal(isRestartableReadOnlyRequestType('exportFilterResultToFile'), false) assert.equal(isRestartableReadOnlyRequestType('unknownFutureWorkerRequest'), false) diff --git a/packages/node-runtime/src/import/message-deduplicator.ts b/packages/node-runtime/src/import/message-deduplicator.ts index d73eb4e9..511f87ca 100644 --- a/packages/node-runtime/src/import/message-deduplicator.ts +++ b/packages/node-runtime/src/import/message-deduplicator.ts @@ -35,8 +35,8 @@ export function createMessageDedupState( } /** - * Canonical file-import dedup rule: prefer stable platform message IDs and - * fall back to timestamp + sender platform ID + type + normalized content + reply target. + * Canonical import dedup rule shared by file imports and JSON Push: prefer stable platform + * message IDs and fall back to timestamp + sender platform ID + type + normalized content + reply target. */ export function registerMessageAndCheckDuplicate(message: DedupMessage, state: MessageDedupState): boolean { const key = generateFallbackMessageKey(message) diff --git a/packages/node-runtime/src/import/session-id.ts b/packages/node-runtime/src/import/session-id.ts new file mode 100644 index 00000000..3e3dc654 --- /dev/null +++ b/packages/node-runtime/src/import/session-id.ts @@ -0,0 +1,5 @@ +const SAFE_IMPORT_SESSION_ID_RE = /^[A-Za-z0-9_@-][A-Za-z0-9._@-]{0,127}$/ + +export function isValidImportSessionId(sessionId: string): boolean { + return SAFE_IMPORT_SESSION_ID_RE.test(sessionId) && !sessionId.includes('..') +} diff --git a/packages/node-runtime/src/services/push-importer.test.ts b/packages/node-runtime/src/services/push-importer.test.ts index d074819c..1efed946 100644 --- a/packages/node-runtime/src/services/push-importer.test.ts +++ b/packages/node-runtime/src/services/push-importer.test.ts @@ -4,6 +4,7 @@ import os from 'node:os' import path from 'node:path' import test from 'node:test' import { DatabaseManager } from '../database-manager' +import { raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta, type RuntimeIdentity } from '../data-dir-compat' import { withDataDirImportLock } from '../import/import-lock' import type { PushImportPayload } from './push-importer' import { pushImport } from './push-importer' @@ -15,7 +16,7 @@ function makeTempDir(): string { return fs.mkdtempSync(path.join(baseDir, 'chatlab-push-import-')) } -function createDatabaseManager(rootDir: string): DatabaseManager { +function createDatabaseManager(rootDir: string, runtime?: RuntimeIdentity): DatabaseManager { return new DatabaseManager( { getSystemDir: () => rootDir, @@ -29,7 +30,7 @@ function createDatabaseManager(rootDir: string): DatabaseManager { getLogsDir: () => path.join(rootDir, 'logs'), getDownloadsDir: () => rootDir, }, - { nativeBinding, allowMissingRuntimeForTests: true } + runtime ? { nativeBinding, runtime } : { nativeBinding, allowMissingRuntimeForTests: true } ) } @@ -54,6 +55,81 @@ test('rejects push imports while any writer holds the data-directory import lock assert.equal(fs.existsSync(manager.getDbPath('different-session')), false) }) +test('rejects unsafe session IDs before resolving or opening a database path', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + const manager = createDatabaseManager(root) + + const outcome = await pushImport(manager, '../escape', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Unsafe Session', platform: 'wechat', type: 'private' }, + messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }], + }) + + assert.deepEqual(outcome, { + ok: false, + reason: 'invalid_payload', + message: 'sessionId contains invalid characters', + }) + assert.equal(fs.existsSync(path.join(root, 'escape.db')), false) +}) + +test('raises the data directory compatibility gate after a successful push import', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + const manager = createDatabaseManager(root, { version: '0.25.1', kind: 'cli' }) + + const outcome = await pushImport(manager, 'compatibility-gate', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Compatibility Gate', platform: 'wechat', type: 'private' }, + messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }], + }) + + assert.equal(outcome.ok, true) + const meta = readDataDirCompatibilityMeta(root) + assert.equal(meta?.minRuntimeVersion, '0.25.1') + assert.equal(meta?.dataCompatibilityVersion, 1) + assert.deepEqual(meta?.reasons, ['segment-schema']) +}) + +test('checks data directory compatibility before creating a push-import database', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + raiseDataDirMinRuntimeVersion( + { + getSystemDir: () => root, + getUserDataDir: () => root, + getDatabaseDir: () => path.join(root, 'databases'), + getVectorDir: () => path.join(root, 'vector'), + getAiDataDir: () => path.join(root, 'ai'), + getSettingsDir: () => path.join(root, 'settings'), + getCacheDir: () => path.join(root, 'cache'), + getTempDir: () => path.join(root, 'temp'), + getLogsDir: () => path.join(root, 'logs'), + getDownloadsDir: () => root, + }, + { + minRuntimeVersion: '0.26.0', + dataCompatibilityVersion: 2, + reason: 'future-schema', + runtime: { version: '0.26.0', kind: 'desktop' }, + module: 'future-migration', + } + ) + const manager = createDatabaseManager(root, { version: '0.25.1', kind: 'cli' }) + + await assert.rejects( + () => + pushImport(manager, 'incompatible-data-dir', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Incompatible', platform: 'wechat', type: 'private' }, + messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }], + }), + /requires runtime version 0\.26\.0 or newer/ + ) + assert.equal(fs.existsSync(path.join(root, 'databases', 'incompatible-data-dir.db')), false) +}) + test('deduplicates initial push batch when a platform-id message is repeated without platform id', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -200,6 +276,76 @@ test('deduplicates incremental push when a later platform-id message matches an } }) +test('keeps distinct platform messages even when timestamp, sender and content match', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const initialPayload: PushImportPayload = { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Distinct Platform Messages', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { + platformMessageId: 'msg-1', + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'same content', + }, + ], + } + + const initialOutcome = await pushImport(manager, 'distinct-platform-messages', initialPayload) + assert.equal(initialOutcome.ok, true) + + const outcome = await pushImport(manager, 'distinct-platform-messages', { + messages: [ + { + platformMessageId: 'msg-2', + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'same content', + }, + ], + }) + + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.deepEqual(outcome.result.batch, { + receivedCount: 1, + writtenCount: 1, + duplicateCount: 0, + }) + + const db = manager.openRawSessionDatabase('distinct-platform-messages', { readonly: true }) + try { + const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number } + assert.equal(row.count, 2) + } finally { + db.close() + } +}) + +test('creates push-import databases only inside the canonical databases directory', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const outcome = await pushImport(manager, 'canonical-path', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Canonical Path', platform: 'wechat', type: 'private' }, + messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }], + }) + + assert.equal(outcome.ok, true) + assert.equal(fs.existsSync(path.join(tempDir, 'databases', 'canonical-path.db')), true) + assert.equal(fs.existsSync(path.join(tempDir, 'canonical-path.db')), false) +}) + test('rejects non-array messages before applying incremental metadata updates', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) diff --git a/packages/node-runtime/src/services/push-importer.ts b/packages/node-runtime/src/services/push-importer.ts index ee2c36d8..55dff1ee 100644 --- a/packages/node-runtime/src/services/push-importer.ts +++ b/packages/node-runtime/src/services/push-importer.ts @@ -2,22 +2,25 @@ * Push import service — handles POST /api/v1/imports/:sessionId * * Accepts a ChatLab Format JSON payload, creates or appends to a session. - * Dedup: platformMessageId (preferred) or content hash (fallback). + * Dedup: platformMessageId (preferred) or deterministic fallback key. */ import * as fs from 'fs' import { DataDirCompatibilityError } from '../data-dir-compat' -import { - CHAT_DB_SCHEMA, - generateMessageKey, - generateSessionIndex, - generateIncrementalSessionIndex, -} from '@openchatlab/core' +import { CHAT_DB_SCHEMA, generateSessionIndex, generateIncrementalSessionIndex } from '@openchatlab/core' import type { DatabaseAdapter } from '@openchatlab/core' import type { DatabaseManager } from '../database-manager' import { buildFtsIndex, insertFtsEntries } from '../fts' import { writeParseResultToDb } from '../import' import { ImportInProgressError, withDataDirImportLock } from '../import/import-lock' +import { + createMessageDedupState, + generateFallbackMessageKey, + registerMessageAndCheckDuplicate, + type MessageDedupState, +} from '../import/message-deduplicator' +import { appLogger } from '../logging/app-logger' +import { isValidImportSessionId } from '../import/session-id' const SYSTEM_SENDER_ID = 'SYSTEM' const SYSTEM_MEMBER_NAME = '系统消息' @@ -73,6 +76,12 @@ export type PushImportOutcome = | { ok: true; result: PushImportResult } | { ok: false; reason: 'import_in_progress' | 'invalid_payload' | 'import_failed'; message: string } +export interface PushImportExecutionDeps { + getDbPath(sessionId: string): string + openDatabase(sessionId: string, options: { readonly?: boolean; create?: boolean }): DatabaseAdapter + deleteDatabase(sessionId: string): void +} + function validatePayload(payload: PushImportPayload, isNew: boolean): string | null { const messages = payload.messages if (!Array.isArray(messages) || messages.length === 0) return 'messages is required and must be a non-empty array' @@ -142,8 +151,7 @@ function normalizeSenderGroupNickname(sender: string, groupNickname: string | un function writeMessages( db: DatabaseAdapter, messages: PushImportMessage[], - existingPmids: Set, - existingKeys: Set + dedupState: MessageDedupState ): { writtenCount: number duplicateCount: number @@ -167,26 +175,21 @@ function writeMessages( db.transaction(() => { for (const msg of sorted) { - const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null) - if (msg.platformMessageId) { - if (existingPmids.has(msg.platformMessageId)) { - duplicateCount++ - continue - } - if (existingKeys.has(key)) { - duplicateCount++ - continue - } - existingPmids.add(msg.platformMessageId) - // Also register the content hash so a same-content no-pmid copy later - // in this batch is caught by the fallback dedup path. - existingKeys.add(key) - } else { - if (existingKeys.has(key)) { - duplicateCount++ - continue - } - existingKeys.add(key) + if ( + registerMessageAndCheckDuplicate( + { + platformMessageId: msg.platformMessageId, + timestamp: msg.timestamp, + senderPlatformId: msg.sender, + type: msg.type, + content: msg.content ?? null, + replyToMessageId: msg.replyToMessageId, + }, + dedupState + ) + ) { + duplicateCount++ + continue } let memberId = memberIdCache.get(msg.sender) @@ -244,29 +247,25 @@ function fullImport( const allMembers = extraMembers.length > 0 ? [...members, ...extraMembers] : members // Deduplicate within the batch using the same pmid/hash logic as incremental imports. - const pmids = new Set() - const hashKeys = new Set() + const dedupState = createMessageDedupState() let duplicateCount = 0 const dedupedMessages: PushImportMessage[] = [] for (const msg of messages) { - const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null) - if (msg.platformMessageId) { - if (pmids.has(msg.platformMessageId)) { - duplicateCount++ - continue - } - if (hashKeys.has(key)) { - duplicateCount++ - continue - } - pmids.add(msg.platformMessageId) - hashKeys.add(key) - } else { - if (hashKeys.has(key)) { - duplicateCount++ - continue - } - hashKeys.add(key) + if ( + registerMessageAndCheckDuplicate( + { + platformMessageId: msg.platformMessageId, + timestamp: msg.timestamp, + senderPlatformId: msg.sender, + type: msg.type, + content: msg.content ?? null, + replyToMessageId: msg.replyToMessageId, + }, + dedupState + ) + ) { + duplicateCount++ + continue } dedupedMessages.push(msg) } @@ -320,19 +319,34 @@ function incrementalImport( // Load dedup keys const existingPmids = new Set() const existingKeys = new Set() - ;( - db.prepare('SELECT platform_message_id FROM message WHERE platform_message_id IS NOT NULL').all() as Array<{ - platform_message_id: string - }> - ).forEach((r) => existingPmids.add(r.platform_message_id)) - // Load hashes for ALL existing messages (not just those without pmid) so that - // messages previously imported with a platformMessageId are still caught by - // content-hash dedup when the same content arrives without one. - ;( - db - .prepare(`SELECT msg.ts, m.platform_id, msg.content FROM message msg JOIN member m ON msg.sender_id = m.id`) - .all() as Array<{ ts: number; platform_id: string; content: string | null }> - ).forEach((r) => existingKeys.add(generateMessageKey(r.ts, r.platform_id, r.content))) + const existingFallbackOnlyKeys = new Set() + const existingMessages = db + .prepare( + `SELECT msg.ts, m.platform_id, msg.type, msg.content, msg.reply_to_message_id, msg.platform_message_id + FROM message msg + JOIN member m ON msg.sender_id = m.id` + ) + .all() as Array<{ + ts: number + platform_id: string + type: number + content: string | null + reply_to_message_id: string | null + platform_message_id: string | null + }> + for (const message of existingMessages) { + if (message.platform_message_id) existingPmids.add(message.platform_message_id) + const key = generateFallbackMessageKey({ + timestamp: message.ts, + senderPlatformId: message.platform_id, + type: message.type, + content: message.content, + replyToMessageId: message.reply_to_message_id ?? undefined, + }) + existingKeys.add(key) + if (!message.platform_message_id) existingFallbackOnlyKeys.add(key) + } + const dedupState = createMessageDedupState(existingPmids, existingKeys, existingFallbackOnlyKeys) let metaUpdated = false let membersAdded = 0 @@ -387,12 +401,7 @@ function incrementalImport( const preWriteMaxTs = (db.prepare('SELECT MAX(ts) as max_ts FROM message').get() as { max_ts: number | null })?.max_ts ?? 0 - const { writtenCount, duplicateCount, minWrittenTs, ftsEntries } = writeMessages( - db, - payload.messages!, - existingPmids, - existingKeys - ) + const { writtenCount, duplicateCount, minWrittenTs, ftsEntries } = writeMessages(db, payload.messages!, dedupState) if (ftsEntries.length > 0) { try { @@ -423,23 +432,34 @@ function incrementalImport( return { writtenCount, duplicateCount, metaUpdated, membersAdded, membersUpdated } } -// Only allow characters that are safe as a bare filename component. -// Rejects path separators (/ \), dots-only sequences (..), and control chars. -const SAFE_SESSION_ID_RE = /^[a-zA-Z0-9_@-][a-zA-Z0-9_@.-]*$/ - export async function pushImport( dbManager: DatabaseManager, sessionId: string, payload: PushImportPayload ): Promise { - if (!SAFE_SESSION_ID_RE.test(sessionId) || sessionId.includes('..')) { - return { ok: false, reason: 'invalid_payload', message: 'sessionId contains invalid characters' } - } - try { - return await withDataDirImportLock(dbManager.getUserDataDir(), () => - pushImportUnlocked(dbManager, sessionId, payload) - ) + return await withDataDirImportLock(dbManager.getUserDataDir(), async () => { + const outcome = await executePushImportUnlocked( + { + getDbPath: (id) => dbManager.getDbPath(id), + openDatabase: (id, options) => dbManager.openRawSessionDatabase(id, options), + deleteDatabase: (id) => { + dbManager.deleteSessionDatabaseFiles(id) + }, + }, + sessionId, + payload + ) + if (!outcome.ok) return outcome + + try { + dbManager.raiseCurrentChatDbCompatibilityGate() + } catch (error) { + if (outcome.result.created) dbManager.deleteSessionDatabaseFiles(sessionId) + throw error + } + return outcome + }) } catch (error) { if (error instanceof ImportInProgressError) { return { @@ -452,12 +472,20 @@ export async function pushImport( } } -async function pushImportUnlocked( - dbManager: DatabaseManager, +/** + * 执行已加锁的 Push Import 写库流程。调用方必须负责数据目录导入锁和兼容门禁， + * 这样 CLI/Internal 可使用 DatabaseManager，Desktop 则可安全地在 worker 中执行写库。 + */ +export async function executePushImportUnlocked( + deps: PushImportExecutionDeps, sessionId: string, payload: PushImportPayload ): Promise { - const dbPath = dbManager.getDbPath(sessionId) + if (!isValidImportSessionId(sessionId)) { + return { ok: false, reason: 'invalid_payload', message: 'sessionId contains invalid characters' } + } + + const dbPath = deps.getDbPath(sessionId) const isNew = !fs.existsSync(dbPath) const validationError = validatePayload(payload, isNew) @@ -467,7 +495,7 @@ async function pushImportUnlocked( try { if (isNew) { - const db = dbManager.openRawSessionDatabase(sessionId, { create: true }) + const db = deps.openDatabase(sessionId, { create: true }) try { const { writtenCount, duplicateCount, membersAdded } = fullImport( db, @@ -476,8 +504,7 @@ async function pushImportUnlocked( payload.messages! ) const session = queryStats(db) - dbManager.raiseCurrentChatDbCompatibilityGate() - return { + const outcome: PushImportOutcome = { ok: true, result: { sessionId, @@ -487,17 +514,18 @@ async function pushImportUnlocked( updates: { metaUpdated: true, membersAdded, membersUpdated: 0 }, }, } + appLogger.info('push-import', 'Push import completed', outcome.result) + return outcome } finally { db.close() } } - const db = dbManager.openRawSessionDatabase(sessionId, { readonly: false }) + const db = deps.openDatabase(sessionId, { readonly: false }) try { const { writtenCount, duplicateCount, metaUpdated, membersAdded, membersUpdated } = incrementalImport(db, payload) const session = queryStats(db) - dbManager.raiseCurrentChatDbCompatibilityGate() - return { + const outcome: PushImportOutcome = { ok: true, result: { sessionId, @@ -507,6 +535,8 @@ async function pushImportUnlocked( updates: { metaUpdated, membersAdded, membersUpdated }, }, } + appLogger.info('push-import', 'Push import completed', outcome.result) + return outcome } finally { db.close() } @@ -517,11 +547,12 @@ async function pushImportUnlocked( if (isNew) { try { - dbManager.deleteSessionDatabaseFiles(sessionId) + deps.deleteDatabase(sessionId) } catch { /* cleanup best-effort */ } } + appLogger.error('push-import', 'Push import failed', err) return { ok: false, reason: 'import_failed', message: err instanceof Error ? err.message : String(err) } } }