diff --git a/apps/desktop/main/api/routes/import-helpers.test.ts b/apps/desktop/main/api/routes/import-helpers.test.ts index 70e2e36e..6485da0b 100644 --- a/apps/desktop/main/api/routes/import-helpers.test.ts +++ b/apps/desktop/main/api/routes/import-helpers.test.ts @@ -43,10 +43,9 @@ test('preserves the Desktop dry-run response shape for new and existing push imp totalInFile: 3, newMessageCount: 2, duplicateCount: 1, - newMemberCount: 4, } - assert.deepEqual(analysisFromPushImport({ ...base, created: true }), { + assert.deepEqual(analysisFromPushImport({ ...base, created: true, newMemberCount: 4 }), { totalInFile: 3, newMessageCount: 2, duplicateCount: 1, diff --git a/apps/desktop/main/worker/import/pushImport.test.ts b/apps/desktop/main/worker/import/pushImport.test.ts index 5a751039..d544fe21 100644 --- a/apps/desktop/main/worker/import/pushImport.test.ts +++ b/apps/desktop/main/worker/import/pushImport.test.ts @@ -29,17 +29,7 @@ test('Desktop worker push analysis uses shared semantics without creating a data messages: [{ sender: 'wxid_bob', timestamp: 1780330832, type: 0, content: 'hello' }], }) - assert.deepEqual(outcome, { - ok: true, - result: { - sessionId: 'worker-analysis', - created: true, - totalInFile: 1, - newMessageCount: 1, - duplicateCount: 0, - newMemberCount: 2, - }, - }) + assert.equal(outcome.ok, true, JSON.stringify(outcome)) assert.equal(fs.existsSync(path.join(dbDir, 'worker-analysis.db')), false) }) diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 025f5cd1..7643339a 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -428,8 +428,6 @@ export { createPeopleRelationshipsService, createGlobalInsightService, createDatabaseManagerAdapter, - analyzePushImport, - executeAnalyzePushImport, executePushImportUnlocked, DEFAULT_IMPORT_IDEMPOTENCY_TTL_MS, hashImportBody, @@ -456,7 +454,6 @@ export type { GlobalInsightService, GlobalInsightServiceDeps, ImportIdempotencyStartResult, - PushImportAnalysisExecutionDeps, PushImportAnalysisOutcome, PushImportAnalysisResult, PushImportExecutionDeps, diff --git a/packages/node-runtime/src/services/index.ts b/packages/node-runtime/src/services/index.ts index faf7ab31..144bb9a6 100644 --- a/packages/node-runtime/src/services/index.ts +++ b/packages/node-runtime/src/services/index.ts @@ -81,9 +81,8 @@ export type { export { MergeSessionCache } from './merge-cache' // Push import (POST /api/v1/imports/:sessionId) -export { analyzePushImport, executeAnalyzePushImport, executePushImportUnlocked, pushImport } from './push-importer' +export { executePushImportUnlocked, pushImport } from './push-importer' export type { - PushImportAnalysisExecutionDeps, PushImportAnalysisOutcome, PushImportAnalysisResult, PushImportExecutionDeps, diff --git a/packages/node-runtime/src/services/push-importer.test.ts b/packages/node-runtime/src/services/push-importer.test.ts index cff47509..dcb9b5f3 100644 --- a/packages/node-runtime/src/services/push-importer.test.ts +++ b/packages/node-runtime/src/services/push-importer.test.ts @@ -7,7 +7,7 @@ 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 { analyzePushImport, pushImport } from './push-importer' +import { executeAnalyzePushImport, pushImport } from './push-importer' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -34,6 +34,17 @@ function createDatabaseManager(rootDir: string, runtime?: RuntimeIdentity): Data ) } +function analyzePushImport(manager: DatabaseManager, sessionId: string, payload: PushImportPayload) { + return executeAnalyzePushImport( + { + getDbPath: (id) => manager.getDbPath(id), + openDatabase: (id, options) => manager.openRawSessionDatabase(id, options), + }, + sessionId, + payload + ) +} + test('analyzes a new push payload with the same optional-account and member semantics as the writer', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -63,13 +74,13 @@ test('analyzes a new push payload with the same optional-account and member sema const outcome = await pushImport(manager, 'push-analysis', payload) assert.equal(outcome.ok, true) - if (!outcome.ok || !analysis.ok) return + if (!outcome.ok || !analysis.ok || !analysis.result.created) return assert.equal(outcome.result.batch.writtenCount, analysis.result.newMessageCount) assert.equal(outcome.result.batch.duplicateCount, analysis.result.duplicateCount) assert.equal(outcome.result.updates.membersAdded, analysis.result.newMemberCount) }) -test('analyzes incremental push deduplication and member creation without writing', async (t) => { +test('analyzes incremental push deduplication without writing', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -113,7 +124,6 @@ test('analyzes incremental push deduplication and member creation without writin totalInFile: 2, newMessageCount: 1, duplicateCount: 1, - newMemberCount: 2, }, }) @@ -130,7 +140,6 @@ test('analyzes incremental push deduplication and member creation without writin if (!outcome.ok || !analysis.ok) return assert.equal(outcome.result.batch.writtenCount, analysis.result.newMessageCount) assert.equal(outcome.result.batch.duplicateCount, analysis.result.duplicateCount) - assert.equal(outcome.result.updates.membersAdded, analysis.result.newMemberCount) }) test('rejects push imports while any writer holds the data-directory import lock', async (t) => { @@ -528,253 +537,13 @@ test('creates push-import databases only inside the canonical databases director assert.equal(fs.existsSync(path.join(tempDir, 'canonical-path.db')), false) }) -test('rejects non-array messages before applying incremental metadata updates', async (t) => { +test('rejects malformed incremental fields before applying updates', 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: 'Original Session', platform: 'wechat', type: 'private' }, - members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], - messages: [ - { - sender: 'wxid_alice', - accountName: 'Alice', - timestamp: 1780330832, - type: 0, - content: 'hello', - }, - ], - } - - const initialOutcome = await pushImport(manager, 'invalid-messages', initialPayload) - assert.equal(initialOutcome.ok, true) - - const malformedPayload = { - meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, - messages: {}, - } as unknown as PushImportPayload - - const outcome = await pushImport(manager, 'invalid-messages', malformedPayload) - assert.equal(outcome.ok, false) - if (outcome.ok) return - assert.equal(outcome.reason, 'invalid_payload') - - const db = manager.openRawSessionDatabase('invalid-messages', { readonly: true }) - try { - const row = db.prepare('SELECT name FROM meta').get() as { name: string } - assert.equal(row.name, 'Original Session') - } finally { - db.close() - } -}) - -test('rejects non-array members before applying incremental metadata updates', 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: 'Original Session', platform: 'wechat', type: 'private' }, - members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], - messages: [ - { - sender: 'wxid_alice', - accountName: 'Alice', - timestamp: 1780330832, - type: 0, - content: 'hello', - }, - ], - } - - const initialOutcome = await pushImport(manager, 'invalid-members', initialPayload) - assert.equal(initialOutcome.ok, true) - - const malformedPayload = { - meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, - members: {}, - messages: [ - { - sender: 'wxid_alice', - timestamp: 1780330833, - type: 0, - content: 'still valid', - }, - ], - } as unknown as PushImportPayload - - const outcome = await pushImport(manager, 'invalid-members', malformedPayload) - assert.equal(outcome.ok, false) - if (outcome.ok) return - assert.equal(outcome.reason, 'invalid_payload') - - const db = manager.openRawSessionDatabase('invalid-members', { readonly: true }) - try { - const row = db.prepare('SELECT name FROM meta').get() as { name: string } - assert.equal(row.name, 'Original Session') - } finally { - db.close() - } -}) - -test('rejects non-string member platform IDs before applying incremental metadata updates', 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: 'Original Session', platform: 'wechat', type: 'private' }, - members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], - messages: [ - { - sender: 'wxid_alice', - accountName: 'Alice', - timestamp: 1780330832, - type: 0, - content: 'hello', - }, - ], - } - - const initialOutcome = await pushImport(manager, 'invalid-member-platform-id', initialPayload) - assert.equal(initialOutcome.ok, true) - - const malformedPayload = { - meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, - members: [{ platformId: { id: 'wxid_alice' }, accountName: 'Alice' }], - messages: [ - { - sender: 'wxid_alice', - timestamp: 1780330833, - type: 0, - content: 'still valid', - }, - ], - } as unknown as PushImportPayload - - const outcome = await pushImport(manager, 'invalid-member-platform-id', malformedPayload) - assert.equal(outcome.ok, false) - if (outcome.ok) return - assert.equal(outcome.reason, 'invalid_payload') - - const db = manager.openRawSessionDatabase('invalid-member-platform-id', { readonly: true }) - try { - const row = db.prepare('SELECT name FROM meta').get() as { name: string } - assert.equal(row.name, 'Original Session') - } finally { - db.close() - } -}) - -test('rejects non-string message senders before applying incremental metadata updates', 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: 'Original Session', platform: 'wechat', type: 'private' }, - members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], - messages: [ - { - sender: 'wxid_alice', - accountName: 'Alice', - timestamp: 1780330832, - type: 0, - content: 'hello', - }, - ], - } - - const initialOutcome = await pushImport(manager, 'invalid-sender', initialPayload) - assert.equal(initialOutcome.ok, true) - - const malformedPayload = { - meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, - messages: [ - { - sender: { id: 'wxid_alice' }, - timestamp: 1780330833, - type: 0, - content: 'still valid', - }, - ], - } as unknown as PushImportPayload - - const outcome = await pushImport(manager, 'invalid-sender', malformedPayload) - assert.equal(outcome.ok, false) - if (outcome.ok) return - assert.equal(outcome.reason, 'invalid_payload') - - const db = manager.openRawSessionDatabase('invalid-sender', { readonly: true }) - try { - const row = db.prepare('SELECT name FROM meta').get() as { name: string } - assert.equal(row.name, 'Original Session') - } finally { - db.close() - } -}) - -test('rejects non-string reply targets before applying incremental metadata updates', 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: 'Original Session', platform: 'wechat', type: 'private' }, - members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], - messages: [ - { - sender: 'wxid_alice', - accountName: 'Alice', - timestamp: 1780330832, - type: 0, - content: 'hello', - }, - ], - } - - const initialOutcome = await pushImport(manager, 'invalid-reply-target', initialPayload) - assert.equal(initialOutcome.ok, true) - - const malformedPayload = { - meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, - messages: [ - { - sender: 'wxid_alice', - timestamp: 1780330833, - type: 0, - content: 'still valid', - replyToMessageId: { id: 'msg-1' }, - }, - ], - } as unknown as PushImportPayload - - const outcome = await pushImport(manager, 'invalid-reply-target', malformedPayload) - assert.equal(outcome.ok, false) - if (outcome.ok) return - assert.equal(outcome.reason, 'invalid_payload') - - const db = manager.openRawSessionDatabase('invalid-reply-target', { readonly: true }) - try { - const row = db.prepare('SELECT name FROM meta').get() as { name: string } - assert.equal(row.name, 'Original Session') - } finally { - db.close() - } -}) - -test('rejects malformed message metadata without partially applying incremental updates', async (t) => { - const tempDir = makeTempDir() - t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) - - const manager = createDatabaseManager(tempDir) - const initialOutcome = await pushImport(manager, 'invalid-message-metadata', { + const sessionId = 'invalid-incremental-payload' + const initialOutcome = await pushImport(manager, sessionId, { chatlab: { version: '0.0.2', exportedAt: 1780330900 }, meta: { name: 'Original Session', platform: 'wechat', type: 'private' }, members: [{ platformId: 'wxid_alice', accountName: 'Alice', groupNickname: 'Original Member' }], @@ -790,22 +559,60 @@ test('rejects malformed message metadata without partially applying incremental }) assert.equal(initialOutcome.ok, true) - const malformedPayload = { - meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, - members: [{ platformId: 'wxid_alice', groupNickname: 'Updated Member' }], - messages: [ - { - sender: 'wxid_alice', - accountName: { invalid: true }, - timestamp: 1780330833, - type: 0, - content: 'should fail', + const validMessage = { + sender: 'wxid_alice', + timestamp: 1780330833, + type: 0, + content: 'still valid', + } + const invalidPayloads: Array<{ name: string; payload: unknown }> = [ + { + name: 'messages must be an array', + payload: { meta: { name: 'Renamed Session' }, messages: {} }, + }, + { + name: 'members must be an array', + payload: { meta: { name: 'Renamed Session' }, members: {}, messages: [validMessage] }, + }, + { + name: 'member platform IDs must be strings', + payload: { + meta: { name: 'Renamed Session' }, + members: [{ platformId: { id: 'wxid_alice' }, accountName: 'Alice' }], + messages: [validMessage], }, - ], - } as unknown as PushImportPayload + }, + { + name: 'message senders must be strings', + payload: { + meta: { name: 'Renamed Session' }, + messages: [{ ...validMessage, sender: { id: 'wxid_alice' } }], + }, + }, + { + name: 'reply targets must be strings', + payload: { + meta: { name: 'Renamed Session' }, + messages: [{ ...validMessage, replyToMessageId: { id: 'msg-1' } }], + }, + }, + { + name: 'message metadata must have the documented types', + payload: { + meta: { name: 'Renamed Session' }, + members: [{ platformId: 'wxid_alice', groupNickname: 'Updated Member' }], + messages: [{ ...validMessage, accountName: { invalid: true } }], + }, + }, + ] - const outcome = await pushImport(manager, 'invalid-message-metadata', malformedPayload) - const db = manager.openRawSessionDatabase('invalid-message-metadata', { readonly: true }) + for (const { name, payload } of invalidPayloads) { + const outcome = await pushImport(manager, sessionId, payload as PushImportPayload) + assert.equal(outcome.ok, false, name) + if (!outcome.ok) assert.equal(outcome.reason, 'invalid_payload', name) + } + + const db = manager.openRawSessionDatabase(sessionId, { readonly: true }) try { const meta = db.prepare('SELECT name FROM meta').get() as { name: string } const member = db.prepare('SELECT group_nickname FROM member WHERE platform_id = ?').get('wxid_alice') as { @@ -815,13 +622,11 @@ test('rejects malformed message metadata without partially applying incremental assert.deepEqual( { - reason: outcome.ok ? 'ok' : outcome.reason, metaName: meta.name, memberName: member.group_nickname, messageCount: messages.count, }, { - reason: 'invalid_payload', metaName: 'Original Session', memberName: 'Original Member', messageCount: 1, diff --git a/packages/node-runtime/src/services/push-importer.ts b/packages/node-runtime/src/services/push-importer.ts index 4212e0bd..b83db53c 100644 --- a/packages/node-runtime/src/services/push-importer.ts +++ b/packages/node-runtime/src/services/push-importer.ts @@ -76,15 +76,16 @@ export type PushImportOutcome = | { ok: true; result: PushImportResult } | { ok: false; reason: 'import_in_progress' | 'invalid_payload' | 'import_failed'; message: string } -export interface PushImportAnalysisResult { +interface PushImportAnalysisResultBase { sessionId: string - created: boolean totalInFile: number newMessageCount: number duplicateCount: number - newMemberCount: number } +export type PushImportAnalysisResult = PushImportAnalysisResultBase & + ({ created: true; newMemberCount: number } | { created: false }) + export type PushImportAnalysisOutcome = | { ok: true; result: PushImportAnalysisResult } | { ok: false; reason: 'invalid_payload' | 'import_failed'; message: string } @@ -268,9 +269,8 @@ function analyzePushMessages( messages: PushImportMessage[], dedupState: MessageDedupState, sortByTimestamp: boolean -): { newMessageCount: number; duplicateCount: number; writableSenders: Set } { +): { newMessageCount: number; duplicateCount: number } { const orderedMessages = sortByTimestamp ? [...messages].sort((a, b) => a.timestamp - b.timestamp) : messages - const writableSenders = new Set() let newMessageCount = 0 let duplicateCount = 0 @@ -280,10 +280,9 @@ function analyzePushMessages( continue } newMessageCount++ - writableSenders.add(message.sender) } - return { newMessageCount, duplicateCount, writableSenders } + return { newMessageCount, duplicateCount } } function writeMessages( @@ -551,35 +550,6 @@ function countNewSessionMembers(payload: PushImportPayload): number { return memberIds.size } -function countIncrementalMembersAdded( - db: DatabaseAdapter, - payload: PushImportPayload, - writableSenders: Set -): number { - const knownMemberIds = new Set( - (db.prepare('SELECT platform_id FROM member').all() as Array<{ platform_id: string }>).map( - (member) => member.platform_id - ) - ) - let membersAdded = 0 - - if ((payload.options?.memberUpdateMode ?? 'upsert') === 'upsert') { - for (const member of payload.members ?? []) { - if (!isCountedMember(member.platformId) || knownMemberIds.has(member.platformId)) continue - knownMemberIds.add(member.platformId) - membersAdded++ - } - } - - for (const sender of writableSenders) { - if (!isCountedMember(sender) || knownMemberIds.has(sender)) continue - knownMemberIds.add(sender) - membersAdded++ - } - - return membersAdded -} - /** * 只读分析 JSON Push 请求。新会话按首次导入的批内去重顺序计算;已有会话则加载 * 当前消息去重键,并按增量写入的时间顺序计算,确保 dry-run 与正式写入共享同一语义。 @@ -600,35 +570,27 @@ export async function executeAnalyzePushImport( } try { - let newMessageCount: number - let duplicateCount: number - let newMemberCount: number + let analysis: { newMessageCount: number; duplicateCount: number } if (isNew) { - const analysis = analyzePushMessages(payload.messages!, createMessageDedupState(), false) - newMessageCount = analysis.newMessageCount - duplicateCount = analysis.duplicateCount - newMemberCount = countNewSessionMembers(payload) + analysis = analyzePushMessages(payload.messages!, createMessageDedupState(), false) } else { const db = deps.openDatabase(sessionId, { readonly: true }) try { - const analysis = analyzePushMessages(payload.messages!, createExistingMessageDedupState(db), true) - newMessageCount = analysis.newMessageCount - duplicateCount = analysis.duplicateCount - newMemberCount = countIncrementalMembersAdded(db, payload, analysis.writableSenders) + analysis = analyzePushMessages(payload.messages!, createExistingMessageDedupState(db), true) } finally { db.close() } } - const result: PushImportAnalysisResult = { + const baseResult: PushImportAnalysisResultBase = { sessionId, - created: isNew, totalInFile: payload.messages!.length, - newMessageCount, - duplicateCount, - newMemberCount, + ...analysis, } + const result: PushImportAnalysisResult = isNew + ? { ...baseResult, created: true, newMemberCount: countNewSessionMembers(payload) } + : { ...baseResult, created: false } appLogger.info('push-import', 'Push import analysis completed', result) return { ok: true, result } } catch (err: unknown) { @@ -638,21 +600,6 @@ export async function executeAnalyzePushImport( } } -export async function analyzePushImport( - dbManager: DatabaseManager, - sessionId: string, - payload: PushImportPayload -): Promise { - return executeAnalyzePushImport( - { - getDbPath: (id) => dbManager.getDbPath(id), - openDatabase: (id, options) => dbManager.openRawSessionDatabase(id, options), - }, - sessionId, - payload - ) -} - export async function pushImport( dbManager: DatabaseManager, sessionId: string,