From 851166188dd3ff825782e7e00babe8867728558a Mon Sep 17 00:00:00 2001 From: digua Date: Thu, 25 Jun 2026 21:34:40 +0800 Subject: [PATCH] fix(import): harden push and pull import guards Combine the PR review fixes for push import validation, deduplication, system sender handling, and pull import lock scoping into one reviewable change. --- apps/desktop/main/api/routes/import.ts | 5 +- .../src/services/push-importer.test.ts | 511 ++++++++++++++++++ .../src/services/push-importer.ts | 67 ++- packages/sync/src/pull-engine.test.ts | 36 ++ packages/sync/src/pull-engine.ts | 9 +- 5 files changed, 607 insertions(+), 21 deletions(-) create mode 100644 packages/node-runtime/src/services/push-importer.test.ts diff --git a/apps/desktop/main/api/routes/import.ts b/apps/desktop/main/api/routes/import.ts index 51817b98..7cfe55a4 100644 --- a/apps/desktop/main/api/routes/import.ts +++ b/apps/desktop/main/api/routes/import.ts @@ -110,7 +110,10 @@ function idempotencyFail(key: string | undefined): void { idempotencyCache.delete(key) } -export function getImportingStatus(): boolean { +export function getImportingStatus(sessionId?: string): boolean { + if (sessionId) { + return isImporting.has(sessionId) || isImporting.has('__legacy__') + } return isImporting.size > 0 } diff --git a/packages/node-runtime/src/services/push-importer.test.ts b/packages/node-runtime/src/services/push-importer.test.ts new file mode 100644 index 00000000..fe03ca2a --- /dev/null +++ b/packages/node-runtime/src/services/push-importer.test.ts @@ -0,0 +1,511 @@ +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 { DatabaseManager } from '../database-manager' +import type { PushImportPayload } from './push-importer' +import { pushImport } from './push-importer' + +const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') + +function makeTempDir(): string { + const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() + return fs.mkdtempSync(path.join(baseDir, 'chatlab-push-import-')) +} + +function createDatabaseManager(rootDir: string): DatabaseManager { + return new DatabaseManager( + { + getSystemDir: () => rootDir, + getUserDataDir: () => rootDir, + getDatabaseDir: () => path.join(rootDir, 'databases'), + getVectorDir: () => path.join(rootDir, 'vector'), + getAiDataDir: () => path.join(rootDir, 'ai'), + getSettingsDir: () => path.join(rootDir, 'settings'), + getCacheDir: () => path.join(rootDir, 'cache'), + getTempDir: () => path.join(rootDir, 'temp'), + getLogsDir: () => path.join(rootDir, 'logs'), + getDownloadsDir: () => rootDir, + }, + { nativeBinding, allowMissingRuntimeForTests: true } + ) +} + +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 })) + + const manager = createDatabaseManager(tempDir) + const payload: PushImportPayload = { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Push Import Session', 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', + }, + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'same content', + }, + ], + } + + const outcome = await pushImport(manager, 'initial-dedup', payload) + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.deepEqual(outcome.result.batch, { + receivedCount: 2, + writtenCount: 1, + duplicateCount: 1, + }) + + const db = manager.openRawSessionDatabase('initial-dedup', { readonly: true }) + try { + const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number } + assert.equal(row.count, 1) + } finally { + db.close() + } +}) + +test('deduplicates initial push batch when a no-platform-id message is repeated with platform id', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const payload: PushImportPayload = { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Push Import Session', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'same content', + }, + { + platformMessageId: 'msg-1', + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'same content', + }, + ], + } + + const outcome = await pushImport(manager, 'initial-reverse-dedup', payload) + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.deepEqual(outcome.result.batch, { + receivedCount: 2, + writtenCount: 1, + duplicateCount: 1, + }) + + const db = manager.openRawSessionDatabase('initial-reverse-dedup', { readonly: true }) + try { + const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number } + assert.equal(row.count, 1) + } finally { + db.close() + } +}) + +test('deduplicates incremental push when a later platform-id message matches an existing content hash', 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: 'Push Import Session', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'same content', + }, + ], + } + + const initialOutcome = await pushImport(manager, 'incremental-pmid-hash-dedup', initialPayload) + assert.equal(initialOutcome.ok, true) + + const duplicatePayload: PushImportPayload = { + messages: [ + { + platformMessageId: 'msg-1', + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'same content', + }, + ], + } + + const outcome = await pushImport(manager, 'incremental-pmid-hash-dedup', duplicatePayload) + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.deepEqual(outcome.result.batch, { + receivedCount: 1, + writtenCount: 0, + duplicateCount: 1, + }) + + const db = manager.openRawSessionDatabase('incremental-pmid-hash-dedup', { readonly: true }) + try { + const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number } + assert.equal(row.count, 1) + } finally { + db.close() + } +}) + +test('rejects non-array messages 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-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('treats auto-created SYSTEM sender as a system member during initial import', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const payload: PushImportPayload = { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'System Session', platform: 'wechat', type: 'group' }, + messages: [ + { + sender: 'SYSTEM', + timestamp: 1780330832, + type: 80, + content: 'Alice joined the group', + }, + ], + } + + const outcome = await pushImport(manager, 'system-sender', payload) + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.equal(outcome.result.session.memberCount, 0) + + const db = manager.openRawSessionDatabase('system-sender', { readonly: true }) + try { + const member = db.prepare('SELECT platform_id, account_name FROM member').get() as { + platform_id: string + account_name: string + } + assert.deepEqual(member, { + platform_id: 'SYSTEM', + account_name: '系统消息', + }) + } finally { + db.close() + } +}) + +test('keeps SYSTEM member canonical during incremental member upserts', 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: 'System Session', platform: 'wechat', type: 'group' }, + messages: [ + { + sender: 'SYSTEM', + timestamp: 1780330832, + type: 80, + content: 'Alice joined the group', + }, + ], + } + + const initialOutcome = await pushImport(manager, 'system-member-upsert', initialPayload) + assert.equal(initialOutcome.ok, true) + + const incrementalPayload: PushImportPayload = { + members: [{ platformId: 'SYSTEM', accountName: 'Bot' }], + messages: [ + { + sender: 'SYSTEM', + timestamp: 1780330833, + type: 80, + content: 'Bob joined the group', + }, + ], + } + + const outcome = await pushImport(manager, 'system-member-upsert', incrementalPayload) + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.equal(outcome.result.session.memberCount, 0) + + const db = manager.openRawSessionDatabase('system-member-upsert', { readonly: true }) + try { + const member = db.prepare('SELECT platform_id, account_name FROM member').get() as { + platform_id: string + account_name: string + } + assert.deepEqual(member, { + platform_id: 'SYSTEM', + account_name: '系统消息', + }) + } finally { + db.close() + } +}) diff --git a/packages/node-runtime/src/services/push-importer.ts b/packages/node-runtime/src/services/push-importer.ts index cdec716f..c775ef35 100644 --- a/packages/node-runtime/src/services/push-importer.ts +++ b/packages/node-runtime/src/services/push-importer.ts @@ -20,6 +20,8 @@ import { writeParseResultToDb } from '../import' // Per-session lock: concurrent imports to different sessions are fine (separate DB files). const importInProgress = new Set() +const SYSTEM_SENDER_ID = 'SYSTEM' +const SYSTEM_MEMBER_NAME = '系统消息' export interface PushImportMessage { sender: string @@ -74,7 +76,7 @@ export type PushImportOutcome = function validatePayload(payload: PushImportPayload, isNew: boolean): string | null { const messages = payload.messages - if (!messages || messages.length === 0) return 'messages is required and must contain at least one message' + if (!Array.isArray(messages) || messages.length === 0) return 'messages is required and must be a non-empty array' if (isNew) { if (!payload.chatlab) return 'chatlab is required for new sessions' @@ -93,15 +95,17 @@ function validatePayload(payload: PushImportPayload, isNew: boolean): string | n return `options.memberUpdateMode must be 'upsert' or 'none'` } + if (payload.members !== undefined && !Array.isArray(payload.members)) return 'members must be an array' if (payload.members) { for (let i = 0; i < payload.members.length; i++) { - if (!payload.members[i].platformId) return `members[${i}].platformId is required` + if (typeof payload.members[i].platformId !== 'string' || payload.members[i].platformId.length === 0) + return `members[${i}].platformId must be a string` } } for (let i = 0; i < messages.length; i++) { const msg = messages[i] - if (!msg.sender) return `messages[${i}].sender is required` + if (typeof msg.sender !== 'string' || msg.sender.length === 0) return `messages[${i}].sender must be a string` if (typeof msg.timestamp !== 'number' || msg.timestamp <= 0) return `messages[${i}].timestamp must be a positive number` if (typeof msg.type !== 'number') return `messages[${i}].type must be a number` @@ -109,6 +113,8 @@ function validatePayload(payload: PushImportPayload, isNew: boolean): string | n return `messages[${i}].content must be a string or null` if (msg.platformMessageId !== undefined && typeof msg.platformMessageId !== 'string') return `messages[${i}].platformMessageId must be a string` + if (msg.replyToMessageId !== undefined && typeof msg.replyToMessageId !== 'string') + return `messages[${i}].replyToMessageId must be a string` } return null @@ -120,10 +126,20 @@ function queryStats(db: DatabaseAdapter): PushImportResult['session'] { first: number | null last: number | null } - const memberRow = db.prepare('SELECT COUNT(*) as cnt FROM member').get() as { cnt: number } + const memberRow = db + .prepare(`SELECT COUNT(*) as cnt FROM member WHERE COALESCE(account_name, '') != ?`) + .get(SYSTEM_MEMBER_NAME) as { cnt: number } return { totalCount: row.total, memberCount: memberRow.cnt, firstTimestamp: row.first, lastTimestamp: row.last } } +function normalizeSenderAccountName(sender: string, accountName: string | undefined): string | undefined { + return sender === SYSTEM_SENDER_ID ? SYSTEM_MEMBER_NAME : accountName +} + +function normalizeSenderGroupNickname(sender: string, groupNickname: string | undefined): string | undefined { + return sender === SYSTEM_SENDER_ID ? SYSTEM_MEMBER_NAME : groupNickname +} + function writeMessages( db: DatabaseAdapter, messages: PushImportMessage[], @@ -152,17 +168,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(generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null)) + existingKeys.add(key) } else { - const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null) if (existingKeys.has(key)) { duplicateCount++ continue @@ -172,7 +192,7 @@ function writeMessages( let memberId = memberIdCache.get(msg.sender) if (!memberId) { - insertMinimalMember.run(msg.sender, msg.accountName || null) + insertMinimalMember.run(msg.sender, normalizeSenderAccountName(msg.sender, msg.accountName) || null) const row = getMemberId.get(msg.sender) as { id: number } | undefined if (row) { memberId = row.id @@ -183,8 +203,8 @@ function writeMessages( const result = insertMsg.run( memberId, - msg.accountName || null, - msg.groupNickname || null, + normalizeSenderAccountName(msg.sender, msg.accountName) || null, + normalizeSenderGroupNickname(msg.sender, msg.groupNickname) || null, msg.timestamp, msg.type, msg.content ?? null, @@ -214,7 +234,11 @@ function fullImport( const extraMembers: PushImportMember[] = [] for (const msg of messages) { if (!knownIds.has(msg.sender)) { - extraMembers.push({ platformId: msg.sender, accountName: msg.accountName }) + extraMembers.push({ + platformId: msg.sender, + accountName: normalizeSenderAccountName(msg.sender, msg.accountName), + groupNickname: normalizeSenderGroupNickname(msg.sender, msg.groupNickname), + }) knownIds.add(msg.sender) } } @@ -226,14 +250,19 @@ function fullImport( 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 { - const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null) if (hashKeys.has(key)) { duplicateCount++ continue @@ -248,11 +277,15 @@ function fullImport( const stats = writeParseResultToDb( db, meta, - allMembers.map((m) => ({ ...m, accountName: m.accountName ?? m.platformId })), + allMembers.map((m) => ({ + ...m, + accountName: normalizeSenderAccountName(m.platformId, m.accountName) ?? m.platformId, + groupNickname: normalizeSenderGroupNickname(m.platformId, m.groupNickname), + })), dedupedMessages.map((m) => ({ senderPlatformId: m.sender, - senderAccountName: m.accountName ?? m.sender, - senderGroupNickname: m.groupNickname, + senderAccountName: normalizeSenderAccountName(m.sender, m.accountName) ?? m.sender, + senderGroupNickname: normalizeSenderGroupNickname(m.sender, m.groupNickname), timestamp: m.timestamp, type: m.type, content: m.content ?? null, @@ -331,10 +364,12 @@ function incrementalImport( db.transaction(() => { for (const m of payload.members!) { const existed = existingMemberIds.has(m.platformId) + const accountName = normalizeSenderAccountName(m.platformId, m.accountName) + const groupNickname = normalizeSenderGroupNickname(m.platformId, m.groupNickname) upsertMember.run( m.platformId, - m.accountName || null, - m.groupNickname || null, + accountName || null, + groupNickname || null, m.avatar || null, m.roles ? JSON.stringify(m.roles) : '[]' ) diff --git a/packages/sync/src/pull-engine.test.ts b/packages/sync/src/pull-engine.test.ts index 4ecc2fb8..cf81fdc7 100644 --- a/packages/sync/src/pull-engine.test.ts +++ b/packages/sync/src/pull-engine.test.ts @@ -69,6 +69,7 @@ function createEngine(options: { files: string[] importResult: Awaited> dataSource: DataSource + isImporting?: (sessionId: string | undefined) => boolean fetchParams?: FetchParams[] sessionUpdates?: Array<{ sessionId: string; updates: Partial }> pullResults?: Array<{ status: 'success' | 'error'; detail: string }> @@ -109,6 +110,7 @@ function createEngine(options: { importer, notifier, dsManager: dsManager as any, + isImporting: options.isImporting, }) } @@ -146,6 +148,40 @@ describe('PullEngine', () => { assert.equal(importCount, 1) }) + it('does not skip pulls when a different local session is importing', async () => { + const session = createSession() + session.targetSessionId = 'local_session_b' + const dataSource = createDataSource() + dataSource.sessions = [session] + const page = writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + messages: [{ sender: 'u1', timestamp: 101, type: 0, content: 'hi' }], + sync: { hasMore: false, nextSince: 101 }, + }) + const checkedSessionIds: Array = [] + const engine = createEngine({ + files: [page], + dataSource, + isImporting: (sessionId) => { + checkedSessionIds.push(sessionId) + return sessionId === 'local_session_a' + }, + importResult: { + success: true, + newMessageCount: 1, + sessionId: session.targetSessionId, + }, + }) + + const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session)) + + assert.equal(result.success, true) + assert.equal(result.newMessageCount, 1) + assert.deepEqual(checkedSessionIds, ['local_session_b']) + }) + it('reports retry import failure instead of marking the pull successful', async () => { const session = createSession() const dataSource = createDataSource() diff --git a/packages/sync/src/pull-engine.ts b/packages/sync/src/pull-engine.ts index c3fbe9fd..ad3d6356 100644 --- a/packages/sync/src/pull-engine.ts +++ b/packages/sync/src/pull-engine.ts @@ -156,8 +156,8 @@ export interface PullEngineOptions { notifier: SyncNotifier dsManager: DataSourceManager logger?: SyncLogger - /** Return true when an import is already in progress (skip pull) */ - isImporting?: () => boolean + /** Return true when an import is already in progress for the local session (skip pull) */ + isImporting?: (sessionId: string | undefined) => boolean /** * Called after a pull session completes successfully with the local session ID. * Used for post-import side effects (e.g. applying the platform owner profile). @@ -172,7 +172,7 @@ export class PullEngine { private notifier: SyncNotifier private dsManager: DataSourceManager private logger: SyncLogger - private isImporting: () => boolean + private isImporting: (sessionId: string | undefined) => boolean private onSessionImported?: (localSessionId: string) => void private pullingSourceIds = new Set() private progressMap = new Map() @@ -222,7 +222,8 @@ export class PullEngine { return { success: true, newMessageCount: 0 } } - if (this.isImporting()) { + const importStatusSessionId = sess.targetSessionId || deriveLocalSessionId(ds.baseUrl, sess.remoteSessionId) + if (this.isImporting(importStatusSessionId)) { this.logger.info(`[Pull] Skipping "${sess.name}": import in progress`) return { success: false, newMessageCount: 0, error: 'Import in progress' } }