diff --git a/apps/cli/src/import/stream-import-flow.test.ts b/apps/cli/src/import/stream-import-flow.test.ts index ad44cc23..6510de60 100644 --- a/apps/cli/src/import/stream-import-flow.test.ts +++ b/apps/cli/src/import/stream-import-flow.test.ts @@ -4,7 +4,7 @@ import os from 'node:os' import path from 'node:path' import test from 'node:test' import Database from 'better-sqlite3' -import type { PathProvider } from '@openchatlab/core' +import { CHAT_DB_SCHEMA, type PathProvider } from '@openchatlab/core' import { DatabaseManager, IMPORT_IN_PROGRESS_ERROR_KEY, @@ -81,6 +81,58 @@ function writeBatchChatFile(dir: string, filename: string, groupId: string, cont return filePath } +function writeScopedBatchUpdate(dir: string): string { + const filePath = path.join(dir, 'scoped-update.json') + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 1783840200 }, + meta: { name: 'Private chat', platform: 'qq', type: 'private' }, + members: [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ], + messages: Array.from({ length: 6 }, (_, index) => ({ + sender: index % 2 === 0 ? '20002' : '10001', + accountName: index % 2 === 0 ? 'Peer' : 'Owner', + timestamp: 1783840200 + index, + type: 0, + content: `message ${index}`, + platformMessageId: `message-${index}`, + })), + }) + ) + return filePath +} + +function seedScopedBatchSession(root: string): void { + const db = new Database(path.join(root, 'data', 'databases', 'scoped-target.db'), { nativeBinding }) + db.exec(CHAT_DB_SCHEMA) + db.prepare( + `INSERT INTO meta (name, platform, type, imported_at, schema_version) + VALUES (?, ?, ?, ?, ?)` + ).run('Private chat', 'qq', 'private', 1783840200, 6) + const insertMember = db.prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)') + const peerId = Number(insertMember.run('uid-peer', 'Peer').lastInsertRowid) + const ownerId = Number(insertMember.run('uid-owner', 'Owner').lastInsertRowid) + const insertMessage = db.prepare( + `INSERT INTO message (sender_id, sender_account_name, ts, type, content, platform_message_id) + VALUES (?, ?, ?, ?, ?, ?)` + ) + for (let index = 0; index < 5; index++) { + const isPeer = index % 2 === 0 + insertMessage.run( + isPeer ? peerId : ownerId, + isPeer ? 'Peer' : 'Owner', + 1783840200 + index, + 0, + `message ${index}`, + `__chatlab_message_scope__0__message-${index}` + ) + } + db.close() +} + test('streamImport raises the data directory gate after creating a current-schema database', async () => { const root = makeTempDir() fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) @@ -233,6 +285,32 @@ test('autoImportBatch holds one lock, coalesces the same target, and preserves i assert.deepEqual(row, { messages: 2, gapThreshold: 7200 }) }) +test('autoImportBatch forwards the matched merger scope into incremental deduplication', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) + seedScopedBatchSession(root) + const manager = new DatabaseManager(createPathProvider(root), { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + + const results = await autoImportBatch(manager, [{ id: 'scoped-update', filePath: writeScopedBatchUpdate(root) }]) + + assert.equal(results[0].status, 'success') + assert.equal(results[0].status === 'success' && results[0].result.importMode, 'incremental') + assert.equal(results[0].status === 'success' && results[0].result.newMessageCount, 1) + assert.equal(results[0].status === 'success' && results[0].result.duplicateCount, 5) + + const db = manager.openRawSessionDatabase('scoped-target', { readonly: true }) + const rows = db.prepare('SELECT platform_message_id FROM message ORDER BY ts, id').all() as Array<{ + platform_message_id: string + }> + db.close() + assert.equal(rows.length, 6) + assert.equal(rows.at(-1)?.platform_message_id, '__chatlab_message_scope__0__message-5') +}) + test('autoImportBatch rejects an external writer and cancels work that has not started', async (t) => { const root = makeTempDir() t.after(() => fs.rmSync(root, { recursive: true, force: true })) diff --git a/apps/cli/src/import/stream-import.ts b/apps/cli/src/import/stream-import.ts index e2778e33..d1b83daf 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -244,7 +244,7 @@ async function autoImportUnlocked( updateCompatibilityGate, itemProgress ), - appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress) => + appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress, context) => incrementalImportUnlocked( dbManager, targetSessionId, @@ -252,6 +252,8 @@ async function autoImportUnlocked( { ...sourceFormatOptions, onProgress: itemProgress ?? progressAdapter, + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, }, updateCompatibilityGate ), @@ -316,7 +318,7 @@ export async function autoImportBatch( false, itemProgress ), - appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress) => + appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress, context) => incrementalImportUnlocked( dbManager, targetSessionId, @@ -324,6 +326,8 @@ export async function autoImportBatch( { ...sourceFormatOptions, onProgress: itemProgress, + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, }, false ), @@ -399,10 +403,12 @@ export async function analyzeAutoImport( formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, }), - analyzeAppendSession: (targetSessionId, sourcePath, sourceFormatOptions) => + analyzeAppendSession: (targetSessionId, sourcePath, sourceFormatOptions, context) => sharedAnalyzeIncremental(targetSessionId, sourcePath, buildIncrementalDeps(dbManager, progressAdapter), { formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, }), }, { diff --git a/apps/desktop/main/worker/import/streamImport.ts b/apps/desktop/main/worker/import/streamImport.ts index bda7347c..74508eac 100644 --- a/apps/desktop/main/worker/import/streamImport.ts +++ b/apps/desktop/main/worker/import/streamImport.ts @@ -138,7 +138,7 @@ export async function autoImport( sessionExists: (sessionId) => fs.existsSync(getDbPath(sessionId)), createSession: (sourcePath, sourceFormatOptions, sessionId, itemProgress) => streamImport(sourcePath, requestId, sourceFormatOptions, sessionId, sessionGapThreshold, itemProgress), - appendSession: (sessionId, sourcePath, sourceFormatOptions, itemProgress) => + appendSession: (sessionId, sourcePath, sourceFormatOptions, itemProgress, context) => incrementalImport( sessionId, sourcePath, @@ -146,6 +146,8 @@ export async function autoImport( { formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, }, itemProgress ), @@ -183,7 +185,7 @@ export async function autoImportBatch( sessionExists: (sessionId) => fs.existsSync(getDbPath(sessionId)), createSession: (sourcePath, sourceFormatOptions, sessionId, itemProgress) => streamImport(sourcePath, requestId, sourceFormatOptions, sessionId, sessionGapThreshold, itemProgress), - appendSession: (sessionId, sourcePath, sourceFormatOptions, itemProgress) => + appendSession: (sessionId, sourcePath, sourceFormatOptions, itemProgress, context) => incrementalImport( sessionId, sourcePath, @@ -191,6 +193,8 @@ export async function autoImportBatch( { formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, }, itemProgress ), diff --git a/packages/node-runtime/src/import/auto-import-matcher.test.ts b/packages/node-runtime/src/import/auto-import-matcher.test.ts index ed90be49..f547f70f 100644 --- a/packages/node-runtime/src/import/auto-import-matcher.test.ts +++ b/packages/node-runtime/src/import/auto-import-matcher.test.ts @@ -7,6 +7,8 @@ import { CHAT_DB_SCHEMA } from '@openchatlab/core' import { MessageType, type ParsedMessage } from '@openchatlab/shared-types' import { openBetterSqliteDatabase } from '../better-sqlite3-adapter' import { resolveAutoImportTarget, resolveAutoImportTargetPlan, type AutoImportMatcherDeps } from './auto-import-matcher' +import { autoImportFile } from './auto-importer' +import { incrementalImport } from './incremental-importer' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -49,6 +51,8 @@ function writeChatLabJsonl( timestamp: message.timestamp, type: message.type, content: message.content, + platformMessageId: message.platformMessageId, + replyToMessageId: message.replyToMessageId, })), ] fs.writeFileSync(filePath, `${lines.map((line) => JSON.stringify(line)).join('\n')}\n`, 'utf8') @@ -64,8 +68,9 @@ function seedSession(dbPath: string, meta: SourceMeta, members: SourceMember[], const insertMember = db.prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)') const insertMessage = db.prepare( - `INSERT INTO message (sender_id, sender_account_name, ts, type, content) - VALUES (?, ?, ?, ?, ?)` + `INSERT INTO message ( + sender_id, sender_account_name, ts, type, content, platform_message_id, reply_to_message_id + ) VALUES (?, ?, ?, ?, ?, ?, ?)` ) const memberIds = new Map() for (const member of members) { @@ -75,7 +80,15 @@ function seedSession(dbPath: string, meta: SourceMeta, members: SourceMember[], for (const message of messages) { const senderId = memberIds.get(message.senderPlatformId) assert.ok(senderId, `missing member ${message.senderPlatformId}`) - insertMessage.run(senderId, message.senderAccountName, message.timestamp, message.type, message.content) + insertMessage.run( + senderId, + message.senderAccountName, + message.timestamp, + message.type, + message.content, + message.platformMessageId ?? null, + message.replyToMessageId ?? null + ) } db.close() } @@ -213,6 +226,86 @@ test('matches a unique private session by stable owner and participant ids', asy ) }) +test('keeps the owner role when matching private sessions with the same participants', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const members = [ + { platformId: 'account-a', accountName: 'Account A' }, + { platformId: 'account-b', accountName: 'Account B' }, + ] + const messages = [textMessage('account-b', 1783840110, 'hello')] + const sourceMeta: SourceMeta = { + name: 'Private chat', + platform: 'qq', + type: 'private', + ownerId: 'account-b', + } + + seedSession(path.join(tempDir, 'account-a.db'), { ...sourceMeta, ownerId: 'account-a' }, members, messages) + seedSession(path.join(tempDir, 'account-b.db'), sourceMeta, members, messages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, sourceMeta, members, messages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['account-a', 'account-b'])), { + action: 'incremental', + sessionId: 'account-b', + matchedBy: 'stable-id', + }) +}) + +test('matches a private session by its complete participant set when owner metadata is missing', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { + name: 'Private chat', + platform: 'qq', + type: 'private', + } + const members = [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ] + const messages = [textMessage('20002', 1783840120, 'hello')] + + seedSession(path.join(tempDir, 'existing.db'), meta, members, messages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, members, messages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['existing'])), { + action: 'incremental', + sessionId: 'existing', + matchedBy: 'stable-id', + }) +}) + +test('matches an owned QQ private session when the import source omits owner metadata', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const sourceMeta: SourceMeta = { + name: 'Private chat', + platform: 'qq', + type: 'private', + } + const members = [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ] + const messages = [textMessage('20002', 1783840130, 'hello')] + + seedSession(path.join(tempDir, 'existing.db'), { ...sourceMeta, ownerId: '10001' }, members, messages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, sourceMeta, members, messages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['existing'])), { + action: 'incremental', + sessionId: 'existing', + matchedBy: 'stable-id', + }) +}) + test('falls back to trailing messages when private stable identity has drifted', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -247,6 +340,403 @@ test('falls back to trailing messages when private stable identity has drifted', }) }) +test('matches QQ private exports across UID and UIN drift using platform message ids', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const candidateMeta: SourceMeta = { + name: 'Private chat', + platform: 'qq', + type: 'private', + ownerId: 'uid-owner', + } + const sourceMeta: SourceMeta = { + ...candidateMeta, + ownerId: '10001', + } + const candidateMembers = [ + { platformId: 'uid-owner', accountName: 'Owner' }, + { platformId: 'uid-peer', accountName: 'Peer' }, + ] + const sourceMembers = [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ] + const candidateMessages = Array.from({ length: 5 }, (_, index) => ({ + ...textMessage(index % 2 === 0 ? 'uid-peer' : 'uid-owner', 1783840140 + index, `message ${index}`), + platformMessageId: `message-${index}`, + })) + const sourceMessages = candidateMessages.map((message, index) => ({ + ...message, + senderPlatformId: index % 2 === 0 ? '20002' : '10001', + })) + + seedSession(path.join(tempDir, 'existing.db'), candidateMeta, candidateMembers, candidateMessages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, sourceMeta, sourceMembers, sourceMessages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['existing'])), { + action: 'incremental', + sessionId: 'existing', + matchedBy: 'trailing-messages', + senderPlatformIdMappings: [ + { sourceId: '10001', targetId: 'uid-owner' }, + { sourceId: '20002', targetId: 'uid-peer' }, + ], + }) +}) + +test('auto import reuses existing QQ members across UID and UIN drift', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const candidateMeta: SourceMeta = { + name: 'Private chat', + platform: 'qq', + type: 'private', + ownerId: 'uid-owner', + } + const sourceMeta: SourceMeta = { + name: 'Private chat', + platform: 'qq', + type: 'private', + } + const candidateMembers = [ + { platformId: 'uid-owner', accountName: 'Owner' }, + { platformId: 'uid-peer', accountName: 'Peer' }, + ] + const sourceMembers = [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ] + const candidateMessages = Array.from({ length: 5 }, (_, index) => ({ + ...textMessage(index % 2 === 0 ? 'uid-peer' : 'uid-owner', 1783840140 + index, `message ${index}`), + platformMessageId: `message-${index}`, + })) + const sourceMessages = [ + ...candidateMessages.map((message, index) => ({ + ...message, + senderPlatformId: index % 2 === 0 ? '20002' : '10001', + })), + { + ...textMessage('20002', 1783840145, 'new message'), + platformMessageId: 'message-5', + }, + ] + + seedSession(path.join(tempDir, 'existing.db'), candidateMeta, candidateMembers, candidateMessages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, sourceMeta, sourceMembers, sourceMessages) + + const result = await autoImportFile(sourcePath, { + ...createDeps(tempDir, ['existing']), + sessionExists: (sessionId) => sessionId === 'existing', + createSession: async () => { + throw new Error('must not create a second session') + }, + appendSession: (sessionId, filePath, _formatOptions, onProgress, context) => + incrementalImport( + sessionId, + filePath, + { + openDatabase: (_targetSessionId, readonly = false) => + openBetterSqliteDatabase(path.join(tempDir, 'existing.db'), { readonly, nativeBinding }), + onProgress: onProgress ?? (() => {}), + }, + { + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, + } + ), + }) + + assert.equal(result.success, true) + assert.equal(result.importMode, 'incremental') + assert.equal(result.matchedBy, 'trailing-messages') + assert.equal(result.newMessageCount, 1) + assert.equal(result.duplicateCount, 5) + + const db = openBetterSqliteDatabase(path.join(tempDir, 'existing.db'), { readonly: true, nativeBinding }) + const meta = db.prepare('SELECT owner_id FROM meta').get() as { owner_id: string | null } + const members = db.prepare('SELECT platform_id FROM member ORDER BY platform_id').all() as Array<{ + platform_id: string + }> + const lastSender = db + .prepare( + `SELECT member.platform_id + FROM message + JOIN member ON member.id = message.sender_id + ORDER BY message.ts DESC, message.id DESC + LIMIT 1` + ) + .get() as { platform_id: string } + db.close() + + assert.equal(meta.owner_id, 'uid-owner') + assert.deepEqual( + members.map((member) => member.platform_id), + ['uid-owner', 'uid-peer'] + ) + assert.equal(lastSender.platform_id, 'uid-peer') +}) + +test('does not match sender-agnostic message ids when the member mapping is incomplete', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { name: 'Private chat', platform: 'qq', type: 'private' } + const candidateMembers = [ + { platformId: 'uid-a', accountName: 'A' }, + { platformId: 'uid-b', accountName: 'B' }, + { platformId: 'uid-c', accountName: 'C' }, + ] + const sourceMembers = [ + { platformId: '10001', accountName: 'A' }, + { platformId: '10002', accountName: 'B' }, + { platformId: '10003', accountName: 'C' }, + ] + const candidateMessages = Array.from({ length: 5 }, (_, index) => ({ + ...textMessage('uid-a', 1783840150 + index, `message ${index}`), + platformMessageId: `message-${index}`, + })) + const sourceMessages = candidateMessages.map((message) => ({ + ...message, + senderPlatformId: '10001', + })) + + seedSession(path.join(tempDir, 'existing.db'), meta, candidateMembers, candidateMessages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, sourceMembers, sourceMessages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['existing'])), { + action: 'create', + reason: 'no-match', + }) +}) + +test('does not guess unmatched QQ group members when participants have changed', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { name: 'Group chat', platform: 'qq', type: 'group' } + const candidateMembers = [ + { platformId: 'uid-a', accountName: 'A' }, + { platformId: 'uid-b', accountName: 'B' }, + { platformId: 'uid-c', accountName: 'C' }, + ] + const sourceMembers = [ + { platformId: '10001', accountName: 'A' }, + { platformId: '10002', accountName: 'B' }, + { platformId: '10004', accountName: 'D' }, + ] + const candidateMessages = Array.from({ length: 5 }, (_, index) => ({ + ...textMessage(index % 2 === 0 ? 'uid-a' : 'uid-b', 1783840160 + index, `message ${index}`), + platformMessageId: `message-${index}`, + })) + const sourceMessages = [ + ...candidateMessages.map((message, index) => ({ + ...message, + senderPlatformId: index % 2 === 0 ? '10001' : '10002', + })), + { + ...textMessage('10004', 1783840165, 'new member message'), + platformMessageId: 'message-5', + }, + ] + + seedSession(path.join(tempDir, 'existing.db'), meta, candidateMembers, candidateMessages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, sourceMembers, sourceMessages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['existing'])), { + action: 'create', + reason: 'no-match', + }) +}) + +test('auto import deduplicates raw IDs against the matched merger scope and keeps new references scoped', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { name: 'Private chat', platform: 'qq', type: 'private' } + const members = [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ] + const existingRawMessages = Array.from({ length: 5 }, (_, index) => ({ + ...textMessage(index % 2 === 0 ? '20002' : '10001', 1783840150 + index, `message ${index}`), + platformMessageId: `message-${index}`, + })) + const rawMessages = [ + ...existingRawMessages, + { + ...textMessage('20002', 1783840155, 'new message'), + platformMessageId: 'message-5', + replyToMessageId: 'message-4', + }, + ] + const scopedMessages = existingRawMessages.map((message) => ({ + ...message, + senderPlatformId: message.senderPlatformId === '20002' ? 'uid-peer' : 'uid-owner', + platformMessageId: `__chatlab_message_scope__0__${encodeURIComponent(message.platformMessageId!)}`, + })) + const candidateMembers = [ + { platformId: 'uid-owner', accountName: 'Owner' }, + { platformId: 'uid-peer', accountName: 'Peer' }, + ] + + seedSession(path.join(tempDir, 'merged.db'), meta, candidateMembers, scopedMessages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, members, rawMessages) + + const result = await autoImportFile(sourcePath, { + ...createDeps(tempDir, ['merged']), + sessionExists: (sessionId) => sessionId === 'merged', + createSession: async () => { + throw new Error('must not create a second session') + }, + appendSession: (sessionId, filePath, _formatOptions, onProgress, context) => + incrementalImport( + sessionId, + filePath, + { + openDatabase: (_targetSessionId, readonly = false) => + openBetterSqliteDatabase(path.join(tempDir, 'merged.db'), { readonly, nativeBinding }), + onProgress: onProgress ?? (() => {}), + }, + { + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, + } + ), + }) + + assert.equal(result.success, true) + assert.equal(result.importMode, 'incremental') + assert.equal(result.matchedBy, 'trailing-messages') + assert.equal(result.newMessageCount, 1) + assert.equal(result.duplicateCount, 5) + + const db = openBetterSqliteDatabase(path.join(tempDir, 'merged.db'), { readonly: true, nativeBinding }) + const rows = db + .prepare('SELECT platform_message_id, reply_to_message_id FROM message ORDER BY ts, id') + .all() as Array<{ platform_message_id: string | null; reply_to_message_id: string | null }> + db.close() + + assert.equal(rows.length, 6) + assert.deepEqual(rows.at(-1), { + platform_message_id: '__chatlab_message_scope__0__message-5', + reply_to_message_id: '__chatlab_message_scope__0__message-4', + }) +}) + +test('auto import preserves every existing namespace in an already scoped source', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { name: 'Merged private chats', platform: 'qq', type: 'private' } + const sourceMembers = [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ] + const candidateMembers = [ + { platformId: 'uid-owner', accountName: 'Owner' }, + { platformId: 'uid-peer', accountName: 'Peer' }, + ] + const existingMessages = Array.from({ length: 5 }, (_, index) => ({ + ...textMessage(index % 2 === 0 ? 'uid-peer' : 'uid-owner', 1783840170 + index, `message ${index}`), + platformMessageId: `__chatlab_message_scope__0__message-${index}`, + })) + const sourceMessages = [ + ...existingMessages.map((message, index) => ({ + ...message, + senderPlatformId: index % 2 === 0 ? '20002' : '10001', + })), + { + ...textMessage('20002', 1783840175, 'second namespace'), + platformMessageId: '__chatlab_message_scope__1__message-5', + replyToMessageId: '__chatlab_message_scope__1__message-4', + }, + ] + + seedSession(path.join(tempDir, 'merged.db'), meta, candidateMembers, existingMessages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, sourceMembers, sourceMessages) + + const result = await autoImportFile(sourcePath, { + ...createDeps(tempDir, ['merged']), + sessionExists: (sessionId) => sessionId === 'merged', + createSession: async () => { + throw new Error('must not create a second session') + }, + appendSession: (sessionId, filePath, _formatOptions, onProgress, context) => + incrementalImport( + sessionId, + filePath, + { + openDatabase: (_targetSessionId, readonly = false) => + openBetterSqliteDatabase(path.join(tempDir, 'merged.db'), { readonly, nativeBinding }), + onProgress: onProgress ?? (() => {}), + }, + { + platformMessageIdScope: context?.platformMessageIdScope, + senderPlatformIdMappings: context?.senderPlatformIdMappings, + } + ), + }) + + assert.equal(result.success, true) + assert.equal(result.importMode, 'incremental') + assert.equal(result.newMessageCount, 1) + assert.equal(result.duplicateCount, 5) + + const db = openBetterSqliteDatabase(path.join(tempDir, 'merged.db'), { readonly: true, nativeBinding }) + const rows = db + .prepare('SELECT platform_message_id, reply_to_message_id FROM message ORDER BY ts, id') + .all() as Array<{ platform_message_id: string | null; reply_to_message_id: string | null }> + db.close() + assert.deepEqual(rows.at(-1), { + platform_message_id: '__chatlab_message_scope__1__message-5', + reply_to_message_id: '__chatlab_message_scope__1__message-4', + }) +}) + +test('does not normalize an already scoped source into a different candidate scope', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const meta: SourceMeta = { name: 'Merged private chats', platform: 'qq', type: 'private' } + const candidateMembers = [ + { platformId: 'uid-owner', accountName: 'Owner' }, + { platformId: 'uid-peer', accountName: 'Peer' }, + ] + const sourceMembers = [ + { platformId: '10001', accountName: 'Owner' }, + { platformId: '20002', accountName: 'Peer' }, + ] + const candidateMessages = Array.from({ length: 5 }, (_, index) => ({ + ...textMessage(index % 2 === 0 ? 'uid-peer' : 'uid-owner', 1783840180 + index, `message ${index}`), + platformMessageId: `__chatlab_message_scope__0__message-${index}`, + })) + const sourceMessages = candidateMessages.map((message, index) => ({ + ...message, + senderPlatformId: index % 2 === 0 ? '20002' : '10001', + platformMessageId: message.platformMessageId?.replace( + '__chatlab_message_scope__0__', + '__chatlab_message_scope__1__' + ), + })) + + seedSession(path.join(tempDir, 'merged.db'), meta, candidateMembers, candidateMessages) + const sourcePath = path.join(tempDir, 'source.jsonl') + writeChatLabJsonl(sourcePath, meta, sourceMembers, sourceMessages) + + assert.deepEqual(await resolveAutoImportTarget(sourcePath, createDeps(tempDir, ['merged'])), { + action: 'create', + reason: 'no-match', + }) +}) + test('prefers a unique stable identity over a different trailing-message candidate', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) diff --git a/packages/node-runtime/src/import/auto-import-matcher.ts b/packages/node-runtime/src/import/auto-import-matcher.ts index 70fed819..86c6a4b1 100644 --- a/packages/node-runtime/src/import/auto-import-matcher.ts +++ b/packages/node-runtime/src/import/auto-import-matcher.ts @@ -2,7 +2,8 @@ import { createHash } from 'node:crypto' import { generateMessageKey, getSessionMeta, isChatSessionDb, type DatabaseAdapter } from '@openchatlab/core' import { streamParseFile, type ParsedMeta } from '@openchatlab/parser' import { MessageType } from '@openchatlab/shared-types' -import { normalizeImportTimestamp } from './incremental-importer' +import { normalizeImportTimestamp, type SenderPlatformIdMapping } from './incremental-importer' +import { parsePlatformMessageId } from './message-deduplicator' import type { ImportProgressCallback } from './streaming-importer' const MATCH_WINDOW_SIZE = 5 @@ -11,7 +12,13 @@ export type AutoImportMatchMethod = 'source-session-id' | 'stable-id' | 'trailin export type AutoImportCreateReason = 'no-match' | 'ambiguous' export type AutoImportDecision = - | { action: 'incremental'; sessionId: string; matchedBy: AutoImportMatchMethod } + | { + action: 'incremental' + sessionId: string + matchedBy: AutoImportMatchMethod + platformMessageIdScope?: string + senderPlatformIdMappings?: SenderPlatformIdMapping[] + } | { action: 'create'; reason: AutoImportCreateReason } export interface AutoImportTargetPlan { @@ -27,10 +34,17 @@ export interface AutoImportMatcherDeps { onProgress?: ImportProgressCallback } -function buildPrivateIdentity(ownerId: string | null | undefined, memberIds: Iterable): string | null { - if (!ownerId) return null - const ids = [...new Set(memberIds)].filter((id) => id && id.toLowerCase() !== 'system').sort() - return ids.length > 0 ? `${ownerId}\0${ids.join('\0')}` : null +function buildPrivateIdentity( + platform: string, + ownerId: string | null | undefined, + memberIds: Iterable +): string | null { + const sortedIds = [...new Set(memberIds)].filter((id) => id && id.toLowerCase() !== 'system').sort() + const validOwnerId = ownerId && ownerId.toLowerCase() !== 'system' ? ownerId : null + if (validOwnerId) { + return sortedIds.length > 0 ? `owner\0${validOwnerId}\0members\0${sortedIds.join('\0')}` : null + } + return platform === 'qq' && sortedIds.length >= 2 ? `members\0${sortedIds.join('\0')}` : null } function isBusinessMessage(type: number, content: string | null): boolean { @@ -40,7 +54,19 @@ function isBusinessMessage(type: number, content: string | null): boolean { function addMessageWindow( recentKeys: string[], windows: Set, - message: { timestamp: unknown; senderPlatformId: string; type: number; content: string | null } + recentPlatformIdKeys: string[], + platformIdWindows: Set, + platformIdSenders: Map, + recentExactPlatformIdKeys: string[], + exactPlatformIdWindows: Set, + exactPlatformIdSenders: Map, + message: { + platformMessageId?: string + timestamp: unknown + senderPlatformId: string + type: number + content: string | null + } ): void { if (!message.senderPlatformId || !isBusinessMessage(message.type, message.content)) return const timestamp = normalizeImportTimestamp(message.timestamp) @@ -49,6 +75,129 @@ function addMessageWindow( recentKeys.push(generateMessageKey(timestamp, message.senderPlatformId, message.content)) if (recentKeys.length > MATCH_WINDOW_SIZE) recentKeys.shift() if (recentKeys.length === MATCH_WINDOW_SIZE) windows.add(recentKeys.join('')) + + const platformIdKey = generatePlatformMessageKey(message, timestamp) + const exactPlatformIdKey = generatePlatformMessageKey(message, timestamp, false) + if (!platformIdKey) { + recentPlatformIdKeys.length = 0 + recentExactPlatformIdKeys.length = 0 + return + } + recordPlatformMessageSender(platformIdSenders, platformIdKey, message.senderPlatformId) + recordPlatformMessageSender(exactPlatformIdSenders, exactPlatformIdKey ?? platformIdKey, message.senderPlatformId) + recentPlatformIdKeys.push(platformIdKey) + if (recentPlatformIdKeys.length > MATCH_WINDOW_SIZE) recentPlatformIdKeys.shift() + if (recentPlatformIdKeys.length === MATCH_WINDOW_SIZE) platformIdWindows.add(recentPlatformIdKeys.join('')) + + recentExactPlatformIdKeys.push(exactPlatformIdKey ?? platformIdKey) + if (recentExactPlatformIdKeys.length > MATCH_WINDOW_SIZE) recentExactPlatformIdKeys.shift() + if (recentExactPlatformIdKeys.length === MATCH_WINDOW_SIZE) { + exactPlatformIdWindows.add(recentExactPlatformIdKeys.join('')) + } +} + +function recordPlatformMessageSender( + senders: Map, + messageKey: string, + senderPlatformId: string +): void { + if (!senders.has(messageKey)) { + senders.set(messageKey, senderPlatformId) + } else if (senders.get(messageKey) !== senderPlatformId) { + senders.set(messageKey, null) + } +} + +function isUsablePlatformId(id: string | null | undefined): id is string { + return Boolean(id && id.toLowerCase() !== 'system') +} + +function deriveSenderPlatformIdMappings( + sourceMemberIds: Iterable, + targetMemberIds: Iterable, + sourceSenders: string[], + targetSenders: string[], + sourceOwnerId?: string | null, + targetOwnerId?: string | null, + allowSingleRemainingPair = false +): SenderPlatformIdMapping[] | null { + const sourceIds = new Set([...sourceMemberIds].filter(isUsablePlatformId)) + const targetIds = new Set([...targetMemberIds].filter(isUsablePlatformId)) + if (isUsablePlatformId(sourceOwnerId)) sourceIds.add(sourceOwnerId) + if (isUsablePlatformId(targetOwnerId)) targetIds.add(targetOwnerId) + + const forward = new Map() + const reverse = new Map() + let hasDrift = false + const addPair = (sourceId: string, targetId: string): boolean => { + if (!sourceIds.has(sourceId) || !targetIds.has(targetId)) return false + const existingTarget = forward.get(sourceId) + const existingSource = reverse.get(targetId) + if ((existingTarget && existingTarget !== targetId) || (existingSource && existingSource !== sourceId)) { + return false + } + forward.set(sourceId, targetId) + reverse.set(targetId, sourceId) + hasDrift ||= sourceId !== targetId + return true + } + + if ( + isUsablePlatformId(sourceOwnerId) && + isUsablePlatformId(targetOwnerId) && + !addPair(sourceOwnerId, targetOwnerId) + ) { + return null + } + for (let index = 0; index < sourceSenders.length; index++) { + if (!addPair(sourceSenders[index], targetSenders[index])) return null + } + + if (!hasDrift) return [] + + for (const sourceId of sourceIds) { + if (!forward.has(sourceId) && targetIds.has(sourceId) && !reverse.has(sourceId)) { + addPair(sourceId, sourceId) + } + } + + const remainingSourceIds = [...sourceIds].filter((id) => !forward.has(id)) + const remainingTargetIds = [...targetIds].filter((id) => !reverse.has(id)) + // In a two-party private chat, one proven participant leaves only one possible peer. + // Group membership can change, so cardinality alone must not pair joined and departed members. + if (allowSingleRemainingPair && remainingSourceIds.length === 1 && remainingTargetIds.length === 1) { + addPair(remainingSourceIds[0], remainingTargetIds[0]) + } + if ([...sourceIds].some((id) => !forward.has(id))) return null + + return [...forward] + .filter(([sourceId, targetId]) => sourceId !== targetId) + .map(([sourceId, targetId]) => ({ sourceId, targetId })) +} + +function getCommonPlatformMessageIdScope(ids: Array): string | undefined { + let commonScope: string | undefined + for (const id of ids) { + if (!id) return undefined + const { scope } = parsePlatformMessageId(id) + if (!scope || (commonScope && scope !== commonScope)) return undefined + commonScope = scope + } + return commonScope +} + +function generatePlatformMessageKey( + message: { platformMessageId?: string; type: number; content: string | null }, + timestamp: number, + normalizeScope = true +): string | null { + if (!message.platformMessageId) return null + const parsedId = parsePlatformMessageId(message.platformMessageId) + return generateMessageKey( + timestamp, + normalizeScope ? parsedId.rawId : message.platformMessageId, + JSON.stringify([message.type, message.content || null]) + ) } export async function resolveAutoImportTarget( @@ -67,7 +216,14 @@ export async function resolveAutoImportTargetPlan( let sourceMeta: ParsedMeta | null = null const sourceMemberIds = new Set() const sourceWindows = new Set() + const sourcePlatformIdWindows = new Set() + const sourcePlatformIdSenders = new Map() + const sourceExactPlatformIdWindows = new Set() + const sourceExactPlatformIdSenders = new Map() const recentSourceKeys: string[] = [] + const recentSourcePlatformIdKeys: string[] = [] + const recentSourceExactPlatformIdKeys: string[] = [] + let sourceHasScopedPlatformMessageIds = false const { formatId, ...parserOptions } = formatOptions ?? {} await streamParseFile( @@ -83,8 +239,22 @@ export async function resolveAutoImportTargetPlan( }, onMessageBatch: (messages) => { for (const message of messages) { + sourceHasScopedPlatformMessageIds ||= Boolean( + (message.platformMessageId && parsePlatformMessageId(message.platformMessageId).scope) || + (message.replyToMessageId && parsePlatformMessageId(message.replyToMessageId).scope) + ) sourceMemberIds.add(message.senderPlatformId) - addMessageWindow(recentSourceKeys, sourceWindows, message) + addMessageWindow( + recentSourceKeys, + sourceWindows, + recentSourcePlatformIdKeys, + sourcePlatformIdWindows, + sourcePlatformIdSenders, + recentSourceExactPlatformIdKeys, + sourceExactPlatformIdWindows, + sourceExactPlatformIdSenders, + message + ) } }, }, @@ -94,7 +264,13 @@ export async function resolveAutoImportTargetPlan( if (!sourceMeta) throw new Error('Import source did not provide metadata') const meta = sourceMeta as ParsedMeta - const privateIdentity = meta.type === 'private' ? buildPrivateIdentity(meta.ownerId, sourceMemberIds) : null + const sourceHasValidOwner = Boolean(meta.ownerId && meta.ownerId.toLowerCase() !== 'system') + // QCE sources intentionally omit owner metadata because their UID/UIN namespaces + // cannot be mapped reliably. Match those sources by participants even when an + // owner profile has since populated the existing session's owner_id. + const matchQqPrivateByMembers = meta.platform === 'qq' && meta.type === 'private' && !sourceHasValidOwner + const privateIdentity = + meta.type === 'private' ? buildPrivateIdentity(meta.platform, meta.ownerId, sourceMemberIds) : null const hasStableIdentity = Boolean(meta.groupId || privateIdentity) const stableIdentity = meta.groupId ? `group\0${meta.groupId}` @@ -106,7 +282,11 @@ export async function resolveAutoImportTargetPlan( : null const stableMatches: string[] = [] - const trailingMatches: string[] = [] + const trailingMatches: Array<{ + sessionId: string + platformMessageIdScope?: string + senderPlatformIdMappings?: SenderPlatformIdMapping[] + }> = [] for (const sessionId of deps.listSessionIds()) { let db: DatabaseAdapter | null = null try { @@ -118,14 +298,16 @@ export async function resolveAutoImportTargetPlan( if (candidate?.platform !== meta.platform || candidate.type !== meta.type) continue const groupMatches = Boolean(meta.groupId && candidate.groupId === meta.groupId) - const memberRows = privateIdentity - ? (db.prepare("SELECT platform_id FROM member WHERE LOWER(platform_id) != 'system'").all() as Array<{ - platform_id: string - }>) - : [] + const memberRows = + privateIdentity || sourcePlatformIdWindows.size > 0 + ? (db.prepare("SELECT platform_id FROM member WHERE LOWER(platform_id) != 'system'").all() as Array<{ + platform_id: string + }>) + : [] const candidatePrivateIdentity = privateIdentity ? buildPrivateIdentity( - candidate.ownerId, + candidate.platform, + matchQqPrivateByMembers ? null : candidate.ownerId, memberRows.map((row) => row.platform_id) ) : null @@ -134,10 +316,10 @@ export async function resolveAutoImportTargetPlan( stableMatches.push(sessionId) } - if (sourceWindows.size > 0) { + if (sourceWindows.size > 0 || sourcePlatformIdWindows.size > 0) { const rows = db .prepare( - `SELECT msg.ts, member.platform_id, msg.type, msg.content + `SELECT msg.ts, member.platform_id, msg.type, msg.content, msg.platform_message_id FROM message msg JOIN member ON member.id = msg.sender_id WHERE msg.type NOT IN (?, ?) @@ -150,14 +332,73 @@ export async function resolveAutoImportTargetPlan( platform_id: string type: number content: string + platform_message_id: string | null }> if (rows.length === MATCH_WINDOW_SIZE) { - const signature = rows - .reverse() - .map((row) => generateMessageKey(row.ts, row.platform_id, row.content)) - .join('') - if (sourceWindows.has(signature)) trailingMatches.push(sessionId) + const orderedRows = rows.reverse() + const signature = orderedRows.map((row) => generateMessageKey(row.ts, row.platform_id, row.content)).join('') + const platformIdKeys = orderedRows.map((row) => + generatePlatformMessageKey( + { + platformMessageId: row.platform_message_id || undefined, + type: row.type, + content: row.content, + }, + row.ts + ) + ) + const platformIdSignature = platformIdKeys.every((key): key is string => key !== null) + ? platformIdKeys.join('') + : null + const exactPlatformIdKeys = orderedRows.map((row) => + generatePlatformMessageKey( + { + platformMessageId: row.platform_message_id || undefined, + type: row.type, + content: row.content, + }, + row.ts, + false + ) + ) + const exactPlatformIdSignature = exactPlatformIdKeys.every((key): key is string => key !== null) + ? exactPlatformIdKeys.join('') + : null + const platformIdsMatch = sourceHasScopedPlatformMessageIds + ? exactPlatformIdSignature !== null && sourceExactPlatformIdWindows.has(exactPlatformIdSignature) + : platformIdSignature !== null && sourcePlatformIdWindows.has(platformIdSignature) + const contentMatches = !sourceHasScopedPlatformMessageIds && sourceWindows.has(signature) + const matchedPlatformIdKeys = sourceHasScopedPlatformMessageIds ? exactPlatformIdKeys : platformIdKeys + const sourceSenderLookup = sourceHasScopedPlatformMessageIds + ? sourceExactPlatformIdSenders + : sourcePlatformIdSenders + const sourceSenders = platformIdsMatch + ? matchedPlatformIdKeys.map((key) => (key ? sourceSenderLookup.get(key) : undefined)) + : [] + const senderPlatformIdMappings = + platformIdsMatch && sourceSenders.every((sender): sender is string => Boolean(sender)) + ? deriveSenderPlatformIdMappings( + sourceMemberIds, + memberRows.map((row) => row.platform_id), + sourceSenders, + orderedRows.map((row) => row.platform_id), + meta.ownerId, + candidate.ownerId, + meta.type === 'private' + ) + : null + const safePlatformIdsMatch = platformIdsMatch && senderPlatformIdMappings !== null + if (contentMatches || safePlatformIdsMatch) { + const platformMessageIdScope = sourceHasScopedPlatformMessageIds + ? undefined + : getCommonPlatformMessageIdScope(orderedRows.map((row) => row.platform_message_id)) + trailingMatches.push({ + sessionId, + ...(platformMessageIdScope ? { platformMessageIdScope } : {}), + ...(safePlatformIdsMatch && senderPlatformIdMappings.length > 0 ? { senderPlatformIdMappings } : {}), + }) + } } } } catch (error) { @@ -172,16 +413,38 @@ export async function resolveAutoImportTargetPlan( if (hasStableIdentity) { if (meta.sourceSessionId && stableMatches.includes(meta.sourceSessionId)) { + const trailingMatch = trailingMatches.find((match) => match.sessionId === meta.sourceSessionId) return { - decision: { action: 'incremental', sessionId: meta.sourceSessionId, matchedBy: 'source-session-id' }, + decision: { + action: 'incremental', + sessionId: meta.sourceSessionId, + matchedBy: 'source-session-id', + ...(trailingMatch?.platformMessageIdScope + ? { platformMessageIdScope: trailingMatch.platformMessageIdScope } + : {}), + ...(trailingMatch?.senderPlatformIdMappings + ? { senderPlatformIdMappings: trailingMatch.senderPlatformIdMappings } + : {}), + }, concurrencyKey: `session:${meta.sourceSessionId}`, exclusive: false, coalesceCreate: false, } } if (stableMatches.length === 1) { + const trailingMatch = trailingMatches.find((match) => match.sessionId === stableMatches[0]) return { - decision: { action: 'incremental', sessionId: stableMatches[0], matchedBy: 'stable-id' }, + decision: { + action: 'incremental', + sessionId: stableMatches[0], + matchedBy: 'stable-id', + ...(trailingMatch?.platformMessageIdScope + ? { platformMessageIdScope: trailingMatch.platformMessageIdScope } + : {}), + ...(trailingMatch?.senderPlatformIdMappings + ? { senderPlatformIdMappings: trailingMatch.senderPlatformIdMappings } + : {}), + }, concurrencyKey: `session:${stableMatches[0]}`, exclusive: false, coalesceCreate: false, @@ -190,18 +453,42 @@ export async function resolveAutoImportTargetPlan( } // 稳定身份缺失、漂移或产生多个候选时，仍只接受唯一的 5 条连续消息重叠。 - if (meta.sourceSessionId && trailingMatches.includes(meta.sourceSessionId)) { + const sourceSessionTrailingMatch = meta.sourceSessionId + ? trailingMatches.find((match) => match.sessionId === meta.sourceSessionId) + : undefined + if (meta.sourceSessionId && sourceSessionTrailingMatch) { return { - decision: { action: 'incremental', sessionId: meta.sourceSessionId, matchedBy: 'source-session-id' }, + decision: { + action: 'incremental', + sessionId: meta.sourceSessionId, + matchedBy: 'source-session-id', + ...(sourceSessionTrailingMatch.platformMessageIdScope + ? { platformMessageIdScope: sourceSessionTrailingMatch.platformMessageIdScope } + : {}), + ...(sourceSessionTrailingMatch.senderPlatformIdMappings + ? { senderPlatformIdMappings: sourceSessionTrailingMatch.senderPlatformIdMappings } + : {}), + }, concurrencyKey: `session:${meta.sourceSessionId}`, exclusive: false, coalesceCreate: false, } } if (trailingMatches.length === 1) { + const [trailingMatch] = trailingMatches return { - decision: { action: 'incremental', sessionId: trailingMatches[0], matchedBy: 'trailing-messages' }, - concurrencyKey: `session:${trailingMatches[0]}`, + decision: { + action: 'incremental', + sessionId: trailingMatch.sessionId, + matchedBy: 'trailing-messages', + ...(trailingMatch.platformMessageIdScope + ? { platformMessageIdScope: trailingMatch.platformMessageIdScope } + : {}), + ...(trailingMatch.senderPlatformIdMappings + ? { senderPlatformIdMappings: trailingMatch.senderPlatformIdMappings } + : {}), + }, + concurrencyKey: `session:${trailingMatch.sessionId}`, exclusive: false, coalesceCreate: false, } diff --git a/packages/node-runtime/src/import/auto-importer.ts b/packages/node-runtime/src/import/auto-importer.ts index 7f4285fb..3d47b034 100644 --- a/packages/node-runtime/src/import/auto-importer.ts +++ b/packages/node-runtime/src/import/auto-importer.ts @@ -6,8 +6,7 @@ import { type AutoImportMatcherDeps, type AutoImportMatchMethod, } from './auto-import-matcher' -import type { IncrementalImportResult } from './incremental-importer' -import type { IncrementalAnalyzeResult } from './incremental-importer' +import type { IncrementalAnalyzeResult, IncrementalImportResult, SenderPlatformIdMapping } from './incremental-importer' import type { AnalyzeNewImportResult, ImportDiagnostics, @@ -23,6 +22,11 @@ export interface AutoImportOptions { resolvedDecision?: AutoImportDecision } +export interface AutoImportAppendContext { + platformMessageIdScope?: string + senderPlatformIdMappings?: SenderPlatformIdMapping[] +} + export interface AutoImportDeps extends AutoImportMatcherDeps { sessionExists(sessionId: string): boolean createSession( @@ -35,7 +39,8 @@ export interface AutoImportDeps extends AutoImportMatcherDeps { sessionId: string, filePath: string, formatOptions?: Record, - onProgress?: ImportProgressCallback + onProgress?: ImportProgressCallback, + context?: AutoImportAppendContext ): Promise resolveTarget?: typeof resolveAutoImportTarget } @@ -46,7 +51,8 @@ export interface AutoImportAnalysisDeps extends AutoImportMatcherDeps { analyzeAppendSession( sessionId: string, filePath: string, - formatOptions?: Record + formatOptions?: Record, + context?: AutoImportAppendContext ): Promise resolveTarget?: typeof resolveAutoImportTarget } @@ -84,7 +90,13 @@ export interface AutoImportAnalysisResult { } type AutoImportPlan = - | { action: 'incremental'; sessionId: string; matchedBy?: AutoImportMatchMethod } + | { + action: 'incremental' + sessionId: string + matchedBy?: AutoImportMatchMethod + platformMessageIdScope?: string + senderPlatformIdMappings?: SenderPlatformIdMapping[] + } | { action: 'create'; sessionId?: string; reason?: AutoImportCreateReason } async function planAutoImport( @@ -95,7 +107,13 @@ async function planAutoImport( if (options.resolvedDecision) { const decision = options.resolvedDecision return decision.action === 'incremental' - ? { action: 'incremental', sessionId: decision.sessionId, matchedBy: decision.matchedBy } + ? { + action: 'incremental', + sessionId: decision.sessionId, + matchedBy: decision.matchedBy, + ...(decision.platformMessageIdScope ? { platformMessageIdScope: decision.platformMessageIdScope } : {}), + ...(decision.senderPlatformIdMappings ? { senderPlatformIdMappings: decision.senderPlatformIdMappings } : {}), + } : { action: 'create', reason: decision.reason } } @@ -114,7 +132,13 @@ async function planAutoImport( options.formatOptions ) return decision.action === 'incremental' - ? { action: 'incremental', sessionId: decision.sessionId, matchedBy: decision.matchedBy } + ? { + action: 'incremental', + sessionId: decision.sessionId, + matchedBy: decision.matchedBy, + ...(decision.platformMessageIdScope ? { platformMessageIdScope: decision.platformMessageIdScope } : {}), + ...(decision.senderPlatformIdMappings ? { senderPlatformIdMappings: decision.senderPlatformIdMappings } : {}), + } : { action: 'create', reason: decision.reason } } @@ -185,7 +209,10 @@ export async function autoImportFile( if (plan.action === 'incremental') { const result = mapIncrementalResult( plan.sessionId, - await deps.appendSession(plan.sessionId, filePath, options.formatOptions, deps.onProgress), + await deps.appendSession(plan.sessionId, filePath, options.formatOptions, deps.onProgress, { + platformMessageIdScope: plan.platformMessageIdScope, + senderPlatformIdMappings: plan.senderPlatformIdMappings, + }), plan.matchedBy ) appLogger.info('import', 'Incremental import completed', { @@ -233,7 +260,10 @@ export async function analyzeAutoImportFile( const plan = await planAutoImport(filePath, deps, options) if (plan.action === 'incremental') { - const analysis = await deps.analyzeAppendSession(plan.sessionId, filePath, options.formatOptions) + const analysis = await deps.analyzeAppendSession(plan.sessionId, filePath, options.formatOptions, { + platformMessageIdScope: plan.platformMessageIdScope, + senderPlatformIdMappings: plan.senderPlatformIdMappings, + }) if (analysis.error) return { success: false, error: analysis.error } const result: AutoImportAnalysisResult = { success: true, diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index 2735f816..3c70894e 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -18,6 +18,7 @@ import { type StreamParseCallbacks, } from '@openchatlab/parser' import { + applyPlatformMessageIdScope, createMessageDedupState, generateFallbackMessageKey, registerMessageAndCheckDuplicate, @@ -28,11 +29,20 @@ import { normalizeSystemMemberName, SYSTEM_MEMBER_NAME, type ImportProgressCallb // ==================== Public interfaces ==================== +export interface SenderPlatformIdMapping { + sourceId: string + targetId: string +} + export interface ImportOptions { metaUpdateMode?: 'patch' | 'none' memberUpdateMode?: 'upsert' | 'none' formatId?: string chatIndex?: number + /** Internal scope selected by automatic matching for merger-namespaced message IDs. */ + platformMessageIdScope?: string + /** Source-to-target sender IDs proven by automatic platform-message matching. */ + senderPlatformIdMappings?: SenderPlatformIdMapping[] } export interface IncrementalAnalyzeResult { @@ -107,6 +117,31 @@ interface ExistingMessageLookup { type CandidateCache = Map +function applyMessageIdScope( + message: T, + scope: string | undefined +): T { + if (!scope) return message + return { + ...message, + platformMessageId: applyPlatformMessageIdScope(message.platformMessageId, scope), + replyToMessageId: applyPlatformMessageIdScope(message.replyToMessageId, scope), + } +} + +function createPlatformIdMapper(mappings: SenderPlatformIdMapping[] | undefined): (platformId: string) => string { + const mapping = new Map(mappings?.map(({ sourceId, targetId }) => [sourceId, targetId])) + return (platformId) => mapping.get(platformId) ?? platformId +} + +function applySenderPlatformIdMapping( + message: T, + mapPlatformId: (platformId: string) => string +): T { + const senderPlatformId = mapPlatformId(message.senderPlatformId) + return senderPlatformId === message.senderPlatformId ? message : { ...message, senderPlatformId } +} + function createExistingMessageLookup(db: DatabaseAdapter): ExistingMessageLookup { const maxMessageId = (db.prepare('SELECT MAX(id) AS max_id FROM message').get() as { max_id: number | null } | undefined)?.max_id ?? 0 @@ -262,6 +297,7 @@ export async function analyzeIncrementalImport( let newMessageCount = 0 let duplicateCount = 0 let platform = formatFeature.platform + const mapPlatformId = createPlatformIdMapper(options?.senderPlatformIdMappings) try { const existingLookup = createExistingMessageLookup(db) @@ -280,13 +316,16 @@ export async function analyzeIncrementalImport( }, onLog: deps.onParserLog, onMessageBatch: (batch) => { + const scopedBatch = batch.map((message) => + applyMessageIdScope(applySenderPlatformIdMapping(message, mapPlatformId), options?.platformMessageIdScope) + ) const existingBatchPlatformIds = findExistingPlatformMessageIds( existingLookup, - batch.map((message) => message.platformMessageId) + scopedBatch.map((message) => message.platformMessageId) ) const candidateCache: CandidateCache = new Map() - for (const msg of batch) { + for (const msg of scopedBatch) { totalInFile++ const timestamp = normalizeImportTimestamp(msg.timestamp) if (timestamp === null) continue @@ -338,6 +377,7 @@ export async function incrementalImport( const metaUpdateMode = options?.metaUpdateMode ?? 'patch' const memberUpdateMode = options?.memberUpdateMode ?? 'upsert' + const mapPlatformId = createPlatformIdMapper(options?.senderPlatformIdMappings) try { const existingLookup = createExistingMessageLookup(db) @@ -420,7 +460,7 @@ export async function incrementalImport( meta.name || '', meta.groupId || '', meta.groupAvatar || '', - meta.ownerId || '', + meta.ownerId ? mapPlatformId(meta.ownerId) : '', Math.floor(Date.now() / 1000) ) metaUpdated = true @@ -428,11 +468,12 @@ export async function incrementalImport( onMembers: (members) => { if (memberUpdateMode === 'none') return for (const m of members) { - const existed = memberIdMap.has(m.platformId) - const accountName = normalizeSystemMemberName(formatFeature.id, m.platformId, m.accountName) - const groupNickname = normalizeSystemMemberName(formatFeature.id, m.platformId, m.groupNickname) + const platformId = mapPlatformId(m.platformId) + const existed = memberIdMap.has(platformId) + const accountName = normalizeSystemMemberName(formatFeature.id, platformId, m.accountName) + const groupNickname = normalizeSystemMemberName(formatFeature.id, platformId, m.groupNickname) upsertMember.run( - m.platformId, + platformId, accountName || null, groupNickname || null, m.aliases ? JSON.stringify(m.aliases) : '[]', @@ -440,8 +481,8 @@ export async function incrementalImport( m.roles ? JSON.stringify(m.roles) : '[]' ) if (!existed) { - const row = getMemberId.get(m.platformId) as { id: number } | undefined - if (row) memberIdMap.set(m.platformId, row.id) + const row = getMemberId.get(platformId) as { id: number } | undefined + if (row) memberIdMap.set(platformId, row.id) membersAdded++ } else { membersUpdated++ @@ -453,13 +494,16 @@ export async function incrementalImport( }, onLog: deps.onParserLog, onMessageBatch: (batch) => { + const scopedBatch = batch.map((message) => + applyMessageIdScope(applySenderPlatformIdMapping(message, mapPlatformId), options?.platformMessageIdScope) + ) const existingBatchPlatformIds = findExistingPlatformMessageIds( existingLookup, - batch.map((message) => message.platformMessageId) + scopedBatch.map((message) => message.platformMessageId) ) const candidateCache: CandidateCache = new Map() - for (const msg of batch) { + for (const msg of scopedBatch) { processedCount++ if (!msg.senderPlatformId) { diff --git a/packages/node-runtime/src/import/index.ts b/packages/node-runtime/src/import/index.ts index 7ec7c651..cdabde8b 100644 --- a/packages/node-runtime/src/import/index.ts +++ b/packages/node-runtime/src/import/index.ts @@ -38,6 +38,7 @@ export type { IncrementalAnalyzeResult, IncrementalImportResult, IncrementalImportDeps, + SenderPlatformIdMapping, } from './incremental-importer' export { isValidImportSessionId } from './session-id' diff --git a/packages/node-runtime/src/import/message-deduplicator.ts b/packages/node-runtime/src/import/message-deduplicator.ts index 511f87ca..1042e96d 100644 --- a/packages/node-runtime/src/import/message-deduplicator.ts +++ b/packages/node-runtime/src/import/message-deduplicator.ts @@ -1,5 +1,7 @@ import { generateMessageKey } from '@openchatlab/core' +const SCOPED_PLATFORM_MESSAGE_ID_PATTERN = /^(__chatlab_message_scope__\d+__)(.*)$/ + export interface MessageDedupState { platformMessageIds: Set /** Fallback keys for every accepted message, regardless of platform ID. */ @@ -17,6 +19,27 @@ export interface DedupMessage { replyToMessageId?: string } +export interface ParsedPlatformMessageId { + rawId: string + scope?: string +} + +export function parsePlatformMessageId(id: string): ParsedPlatformMessageId { + const scoped = SCOPED_PLATFORM_MESSAGE_ID_PATTERN.exec(id) + if (!scoped) return { rawId: id } + + try { + return { rawId: decodeURIComponent(scoped[2]), scope: scoped[1] } + } catch { + return { rawId: id } + } +} + +export function applyPlatformMessageIdScope(id: string | undefined, scope: string | undefined): string | undefined { + if (!id || !scope) return id + return `${scope}${encodeURIComponent(parsePlatformMessageId(id).rawId)}` +} + export function generateFallbackMessageKey(message: Omit): string { const contentKey = generateMessageKey(message.timestamp, message.senderPlatformId, message.content) return JSON.stringify([contentKey, message.type, message.replyToMessageId || null]) diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 7e6829a3..1b17d45e 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -64,6 +64,7 @@ export type { IncrementalAnalyzeResult, IncrementalImportResult, IncrementalImportDeps, + SenderPlatformIdMapping, AutoImportDecision, AutoImportCreateReason, AutoImportMatcherDeps,