diff --git a/packages/http-routes/src/routes/web/merge.test.ts b/packages/http-routes/src/routes/web/merge.test.ts index 9d05bfc3..d79c54d5 100644 --- a/packages/http-routes/src/routes/web/merge.test.ts +++ b/packages/http-routes/src/routes/web/merge.test.ts @@ -81,3 +81,98 @@ test('forwards the session gap threshold when importing merged output', async (t assert.equal(response.json().sessionId, 'merged-session') assert.deepEqual(receivedOptions, { sessionGapThreshold: 7200 }) }) + +test('keeps every distinct platform message when merging a source with its superset', async (t) => { + const rootDir = makeTempDir() + t.after(() => fs.rmSync(rootDir, { recursive: true, force: true })) + + const pathProvider = createPathProvider(rootDir) + const dbManager = new DatabaseManager(pathProvider, { nativeBinding, allowMissingRuntimeForTests: true }) + const mergeSessionCache = new MergeSessionCache(pathProvider, { nativeBinding }) + + const makeMessages = (count: number) => + Array.from({ length: count }, (_, index) => { + const fallbackCollisionIndex = index < 10 ? Math.floor(index / 2) : index + return { + platformMessageId: `message-${index}`, + senderPlatformId: 'qq-user-1', + senderAccountName: 'Alice', + timestamp: 1_700_000_000 + fallbackCollisionIndex, + type: 0 as const, + content: index < 10 ? `same-second-content-${fallbackCollisionIndex}` : `message-content-${index}`, + replyToMessageId: index === 11 ? 'message-10' : undefined, + } + }) + + const createSession = (sessionId: string, messageCount: number): void => { + const db = dbManager.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true }) + db.prepare('INSERT INTO meta (name, platform, type, imported_at) VALUES (?, ?, ?, ?)').run( + 'Source', + 'qq', + 'private', + 1_700_000_000 + ) + const memberResult = db + .prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)') + .run('qq-user-1', 'Alice') + const insertMessage = db.prepare( + `INSERT INTO message ( + sender_id, sender_account_name, ts, type, content, platform_message_id, reply_to_message_id + ) VALUES (?, ?, ?, ?, ?, ?, ?)` + ) + for (const message of makeMessages(messageCount)) { + insertMessage.run( + memberResult.lastInsertRowid, + message.senderAccountName, + message.timestamp, + message.type, + message.content, + message.platformMessageId, + message.replyToMessageId || null + ) + } + db.close() + } + + createSession('older', 100) + createSession('newer', 120) + + const app = Fastify() + registerMergeRoutes(app, { dbManager, mergeSessionCache }) + await app.ready() + t.after(() => app.close()) + + const exportResponse = await app.inject({ + method: 'POST', + url: '/_web/sessions/export-for-merge', + payload: { sessionIds: ['older', 'newer'] }, + }) + assert.equal(exportResponse.statusCode, 200) + const handles = exportResponse.json().handles as Array<{ handle: string }> + + const conflictResponse = await app.inject({ + method: 'POST', + url: '/_web/merge/conflicts', + payload: { handles: handles.map(({ handle }) => handle) }, + }) + assert.equal(conflictResponse.statusCode, 200) + assert.equal(conflictResponse.json().totalMessages, 120) + + const response = await app.inject({ + method: 'POST', + url: '/_web/merge/execute', + payload: { + handles: handles.map(({ handle }) => handle), + outputName: 'Merged', + }, + }) + + assert.equal(response.statusCode, 200) + const messages = response.json().data.messages as Array<{ + platformMessageId?: string + replyToMessageId?: string + }> + assert.equal(messages.length, 120) + assert.equal(new Set(messages.map((message) => message.platformMessageId)).size, 120) + assert.equal(messages.find((message) => message.platformMessageId === 'message-11')?.replyToMessageId, 'message-10') +}) diff --git a/packages/http-routes/src/routes/web/merge.ts b/packages/http-routes/src/routes/web/merge.ts index c67adef2..47399620 100644 --- a/packages/http-routes/src/routes/web/merge.ts +++ b/packages/http-routes/src/routes/web/merge.ts @@ -251,12 +251,14 @@ export function registerMergeRoutes(server: FastifyInstance, ctx: MergeRoutesCon ) writer.writeMessages( exported.messages.map((msg) => ({ + platformMessageId: msg.platformMessageId, senderPlatformId: msg.sender, senderAccountName: msg.accountName, senderGroupNickname: msg.groupNickname, timestamp: msg.timestamp, type: msg.type, content: msg.content, + replyToMessageId: msg.replyToMessageId, })) ) writer.finish() diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index fce35da7..aed6cad1 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -1025,8 +1025,10 @@ export async function streamParseFileInfo( 'INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname, avatar) VALUES (?, ?, ?, ?)' ) const insertMessage = db.prepare( - `INSERT INTO message (sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content) - VALUES (?, ?, ?, ?, ?, ?)` + `INSERT INTO message ( + platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname, + timestamp, type, content, reply_to_message_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` ) let meta: ParsedMeta = { name: 'Unknown', platform: formatFeature.platform, type: 0 as any } @@ -1076,12 +1078,14 @@ export async function streamParseFileInfo( } insertMessage.run( + msg.platformMessageId || null, msg.senderPlatformId, msg.senderAccountName || null, msg.senderGroupNickname || null, msg.timestamp, msg.type, - msg.content || null + msg.content || null, + msg.replyToMessageId || null ) messageCount++ } diff --git a/packages/node-runtime/src/merger/__tests__/merger.test.ts b/packages/node-runtime/src/merger/__tests__/merger.test.ts index 13d075f4..1e9c274f 100644 --- a/packages/node-runtime/src/merger/__tests__/merger.test.ts +++ b/packages/node-runtime/src/merger/__tests__/merger.test.ts @@ -1,17 +1,22 @@ import { describe, it } from 'node:test' import assert from 'node:assert/strict' import Database from 'better-sqlite3' -import { checkConflictsFromSources, buildMergedOutput, serializeChatLabToJsonl } from '../index' +import { + checkConflictsFromSources, + buildMergedOutput, + serializeChatLabToJsonl, + type MergerInputMessage, +} from '../index' import { exportSessionToJson } from '../temp-db' import type { MergerDataSource, MergerSourceMeta } from '../index' -import type { MergerMember, MergerMessage } from '@openchatlab/core' +import type { MergerMember } from '@openchatlab/core' import { CHATLAB_FORMAT_VERSION } from '@openchatlab/shared-types' import { BetterSqliteAdapter } from '../../better-sqlite3-adapter' function createMockSource( meta: MergerSourceMeta, members: MergerMember[], - messages: MergerMessage[] + messages: MergerInputMessage[] ): MergerDataSource { return { getMeta: () => meta, @@ -127,6 +132,214 @@ describe('buildMergedOutput', () => { ) assert.equal(result.chatLabData.meta.platform, 'mixed') }) + + it('keeps same-platform message IDs separate across unrelated private chats', () => { + const first = createMockSource( + { name: 'Alice', platform: 'telegram', type: 'private' }, + [{ platformId: 'owner' }, { platformId: 'alice' }], + [ + { + platformMessageId: '1', + senderPlatformId: 'alice', + timestamp: 100, + type: 0, + content: 'first chat', + }, + ] + ) + const second = createMockSource( + { name: 'Bob', platform: 'telegram', type: 'private' }, + [{ platformId: 'owner' }, { platformId: 'bob' }], + [ + { + platformMessageId: '1', + senderPlatformId: 'bob', + timestamp: 200, + type: 0, + content: 'second chat', + }, + ] + ) + + const result = buildMergedOutput( + [ + { source: first, filename: 'alice.json' }, + { source: second, filename: 'bob.json' }, + ], + 'Merged' + ) + + const conflicts = checkConflictsFromSources([ + { source: first, filename: 'alice.json' }, + { source: second, filename: 'bob.json' }, + ]) + assert.equal(conflicts.totalMessages, 2) + assert.equal(result.chatLabData.messages.length, 2) + assert.equal(new Set(result.chatLabData.messages.map((message) => message.platformMessageId)).size, 2) + }) + + it('deduplicates overlapping private exports when their observed members differ', () => { + const meta: MergerSourceMeta = { name: 'Alice', platform: 'qq', type: 'private' } + const sharedMessage: MergerInputMessage = { + platformMessageId: 'message-1', + senderPlatformId: 'alice', + timestamp: 100, + type: 0, + content: 'shared', + } + const incomingOnly = createMockSource(meta, [{ platformId: 'alice' }], [sharedMessage]) + const bothSides = createMockSource( + meta, + [{ platformId: 'owner' }, { platformId: 'alice' }], + [ + sharedMessage, + { + platformMessageId: 'message-2', + senderPlatformId: 'owner', + timestamp: 200, + type: 0, + content: 'reply', + }, + ] + ) + const sources = [ + { source: incomingOnly, filename: 'old.json' }, + { source: bothSides, filename: 'new.json' }, + ] + + const conflicts = checkConflictsFromSources(sources) + const result = buildMergedOutput(sources, 'Merged') + + assert.equal(conflicts.totalMessages, 2) + assert.equal(result.chatLabData.messages.length, 2) + assert.deepEqual( + result.chatLabData.messages.map((message) => message.platformMessageId), + ['message-1', 'message-2'] + ) + }) + + it('deduplicates the same group when a newer export has a different name and more members', () => { + const oldMeta: MergerSourceMeta = { name: 'Old project', platform: 'qq', type: 'group' } + const newMeta: MergerSourceMeta = { name: 'Renamed project', platform: 'qq', type: 'group' } + const sharedMessage: MergerInputMessage = { + platformMessageId: 'message-1', + senderPlatformId: 'alice', + timestamp: 100, + type: 0, + content: 'shared', + } + const oldExport = createMockSource(oldMeta, [{ platformId: 'alice' }, { platformId: 'bob' }], [sharedMessage]) + const newExport = createMockSource( + newMeta, + [{ platformId: 'alice' }, { platformId: 'bob' }, { platformId: 'carol' }], + [ + sharedMessage, + { + platformMessageId: 'message-2', + replyToMessageId: 'message-1', + senderPlatformId: 'carol', + timestamp: 200, + type: 0, + content: 'new', + }, + ] + ) + const sources = [ + { source: oldExport, filename: 'old.json' }, + { source: newExport, filename: 'new.json' }, + ] + + const conflicts = checkConflictsFromSources(sources) + const result = buildMergedOutput(sources, 'Merged') + + assert.equal(conflicts.totalMessages, 2) + assert.equal(result.chatLabData.messages.length, 2) + assert.deepEqual( + result.chatLabData.messages.map((message) => message.platformMessageId), + ['message-1', 'message-2'] + ) + assert.equal(result.chatLabData.messages[1].replyToMessageId, result.chatLabData.messages[0].platformMessageId) + }) + + it('keeps the same message ID separate across unrelated groups with the same name', () => { + const meta: MergerSourceMeta = { name: 'Project', platform: 'qq', type: 'group' } + const first = createMockSource( + meta, + [{ platformId: 'alice' }], + [ + { + platformMessageId: 'message-1', + senderPlatformId: 'alice', + timestamp: 100, + type: 0, + content: 'first group', + }, + ] + ) + const second = createMockSource( + meta, + [{ platformId: 'bob' }], + [ + { + platformMessageId: 'message-1', + senderPlatformId: 'bob', + timestamp: 200, + type: 0, + content: 'second group', + }, + ] + ) + const sources = [ + { source: first, filename: 'first.json' }, + { source: second, filename: 'second.json' }, + ] + + const conflicts = checkConflictsFromSources(sources) + const result = buildMergedOutput(sources, 'Merged') + + assert.equal(conflicts.totalMessages, 2) + assert.equal(result.chatLabData.messages.length, 2) + assert.equal(new Set(result.chatLabData.messages.map((message) => message.platformMessageId)).size, 2) + }) + + it('backfills a stable ID when an ID-bearing copy bridges a fallback-only message', () => { + const meta: MergerSourceMeta = { name: 'Chat', platform: 'qq', type: 'private' } + const members = [{ platformId: 'owner' }, { platformId: 'alice' }, { platformId: 'bob' }] + const fallbackSource = createMockSource(meta, members, [ + { senderPlatformId: 'alice', timestamp: 100, type: 0, content: 'root' }, + ]) + const idSource = createMockSource(meta, members, [ + { + platformMessageId: 'message-1', + senderPlatformId: 'alice', + timestamp: 100, + type: 0, + content: 'root', + }, + { + platformMessageId: 'message-2', + replyToMessageId: 'message-1', + senderPlatformId: 'bob', + timestamp: 101, + type: 0, + content: 'reply', + }, + ]) + + const result = buildMergedOutput( + [ + { source: fallbackSource, filename: 'old.json' }, + { source: idSource, filename: 'new.json' }, + ], + 'Merged' + ) + + assert.equal(result.chatLabData.messages.length, 2) + const root = result.chatLabData.messages.find((message) => message.content === 'root') + const reply = result.chatLabData.messages.find((message) => message.content === 'reply') + assert.equal(root?.platformMessageId, 'message-1') + assert.equal(reply?.replyToMessageId, root?.platformMessageId) + }) }) describe('serializeChatLabToJsonl', () => { @@ -182,22 +395,28 @@ describe('exportSessionToJson', () => { avatar TEXT ); CREATE TABLE message ( + id INTEGER PRIMARY KEY AUTOINCREMENT, sender_id INTEGER NOT NULL, sender_account_name TEXT, sender_group_nickname TEXT, ts INTEGER NOT NULL, type INTEGER NOT NULL, - content TEXT + content TEXT, + platform_message_id TEXT, + reply_to_message_id TEXT ); INSERT INTO meta (name, platform, type) VALUES ('Test', 'qq', 'group'); INSERT INTO member (id, platform_id, account_name) VALUES (1, 'alice', 'Alice'); - INSERT INTO message (sender_id, sender_account_name, ts, type, content) - VALUES (1, 'Alice', 100, 0, 'Hello'); + INSERT INTO message ( + sender_id, sender_account_name, ts, type, content, platform_message_id, reply_to_message_id + ) VALUES (1, 'Alice', 100, 0, 'Hello', 'message-1', 'message-0'); `) try { const exported = exportSessionToJson(new BetterSqliteAdapter(rawDb)) assert.equal(exported.chatlab.version, CHATLAB_FORMAT_VERSION) + assert.equal(exported.messages[0].platformMessageId, 'message-1') + assert.equal(exported.messages[0].replyToMessageId, 'message-0') } finally { rawDb.close() } diff --git a/packages/node-runtime/src/merger/index.ts b/packages/node-runtime/src/merger/index.ts index d27ede2c..c5ecb109 100644 --- a/packages/node-runtime/src/merger/index.ts +++ b/packages/node-runtime/src/merger/index.ts @@ -7,10 +7,10 @@ import { getCollidingPlatformIds, + getCollidingPlatformIdsFromMessages, normalizePlatformId, detectConflictsInMessages, mergeMergerMembers, - generateMessageKey, type MergerMember, type MergerMessage, type ConflictCheckResult, @@ -18,6 +18,21 @@ import { type MergedMessage, } from '@openchatlab/core' import { CHATLAB_FORMAT_VERSION } from '@openchatlab/shared-types' +import { + createMessageDedupState, + generateFallbackMessageKey, + registerMessageAndCheckDuplicate, +} from '../import/message-deduplicator' + +export interface MergerInputMessage extends MergerMessage { + platformMessageId?: string + replyToMessageId?: string +} + +export interface ChatLabMergedMessage extends MergedMessage { + platformMessageId?: string + replyToMessageId?: string +} // ==================== Data source abstraction ==================== @@ -29,6 +44,113 @@ export interface MergerSourceMeta { groupAvatar?: string } +function createMessageScope(meta: MergerSourceMeta | null, members: MergerMember[], sourceIndex: number): string { + const platform = meta?.platform || 'unknown' + const type = meta?.type || 'unknown' + + if (meta?.groupId) return JSON.stringify([platform, type, 'group', meta.groupId]) + + // A group name and its membership can both collide or change over time. Keep + // the source isolated until exact overlapping messages prove it is the same chat. + if (type === 'group') return JSON.stringify([platform, type, 'source', sourceIndex]) + + const memberIds = [...new Set(members.map((member) => member.platformId))].sort() + if (memberIds.length > 0) { + if (type === 'private') return JSON.stringify([platform, type, 'members', memberIds]) + return JSON.stringify([platform, type, 'name-members', meta?.name || '', memberIds]) + } + + // Without a stable group ID or participant set, sharing message IDs across + // sources could silently merge unrelated chats. Keep those source-local. + return JSON.stringify([platform, type, 'source', sourceIndex]) +} + +function createMessageBridgeFamily(meta: MergerSourceMeta | null): string | undefined { + if (!meta || meta.groupId || (meta.type !== 'group' && meta.type !== 'private')) return undefined + // The exact message signature proves overlap; the editable chat name must + // not prevent two exports of the same chat from sharing a scope. + return JSON.stringify([meta.platform || 'unknown', meta.type]) +} + +function createMessageIdNormalizer( + scopes: Iterable +): (id: string | undefined, scope: string) => string | undefined { + const uniqueScopes = [...new Set(scopes)] + const scopeIndexes = new Map(uniqueScopes.map((scope, index) => [scope, index])) + const namespaceMessageIds = uniqueScopes.length > 1 + + return (id, scope) => { + if (!id || !namespaceMessageIds) return id + return `__chatlab_message_scope__${scopeIndexes.get(scope) ?? 0}__${encodeURIComponent(id)}` + } +} + +interface MessageScopeContext { + sourceIndex: number + messageScope: string + messageBridgeFamily?: string +} + +function createMessageScopeResolver(contexts: Iterable): { + addMessage: (message: MergerInputMessage, context: MessageScopeContext) => void + complete: () => void + resolve: (sourceIndex: number) => string +} { + const contextBySource = new Map() + const parents = new Map() + const firstSourceBySignature = new Map() + + for (const context of contexts) { + contextBySource.set(context.sourceIndex, context) + parents.set(context.sourceIndex, context.sourceIndex) + } + + const find = (sourceIndex: number): number => { + const parent = parents.get(sourceIndex) ?? sourceIndex + if (parent === sourceIndex) return sourceIndex + const root = find(parent) + parents.set(sourceIndex, root) + return root + } + + const union = (firstSourceIndex: number, secondSourceIndex: number): void => { + const firstRoot = find(firstSourceIndex) + const secondRoot = find(secondSourceIndex) + if (firstRoot === secondRoot) return + + const root = Math.min(firstRoot, secondRoot) + parents.set(firstRoot === root ? secondRoot : firstRoot, root) + } + + return { + addMessage(message, context) { + if (!context.messageBridgeFamily || !message.platformMessageId) return + + const fallbackKey = generateFallbackMessageKey({ + timestamp: message.timestamp, + senderPlatformId: message.senderPlatformId, + type: message.type, + content: message.content ?? null, + replyToMessageId: message.replyToMessageId, + }) + const signature = JSON.stringify([context.messageBridgeFamily, message.platformMessageId, fallbackKey]) + const firstSourceIndex = firstSourceBySignature.get(signature) + if (firstSourceIndex === undefined) { + firstSourceBySignature.set(signature, context.sourceIndex) + return + } + union(firstSourceIndex, context.sourceIndex) + }, + complete() { + firstSourceBySignature.clear() + }, + resolve(sourceIndex) { + const rootContext = contextBySource.get(find(sourceIndex)) + return rootContext?.messageScope ?? JSON.stringify(['unknown', 'source', sourceIndex]) + }, + } +} + /** * Abstract data source for merger input. * In Electron, this wraps TempDbReader. Other platforms can implement @@ -38,7 +160,7 @@ export interface MergerDataSource { getMeta(): MergerSourceMeta | null getMembers(): MergerMember[] getMessageCount(): number - streamMessages(batchSize: number, callback: (messages: MergerMessage[]) => void): void + streamMessages(batchSize: number, callback: (messages: MergerInputMessage[]) => void): void } // ==================== Output types ==================== @@ -69,7 +191,7 @@ export interface ChatLabOutput { chatlab: ChatLabHeader meta: ChatLabMeta members: MergedMember[] - messages: MergedMessage[] + messages: ChatLabMergedMessage[] } // ==================== Conflict checking ==================== @@ -77,19 +199,64 @@ export interface ChatLabOutput { export function checkConflictsFromSources( dataSources: Array<{ source: MergerDataSource; filename: string }> ): ConflictCheckResult { - const allMessages: Array<{ msg: MergerMessage; source: string; platform: string }> = [] - - for (const { source, filename } of dataSources) { + const preparedSources = dataSources.map(({ source, filename }, sourceIndex) => { const meta = source.getMeta() - const platform = meta?.platform || 'unknown' + const members = source.getMembers() + return { + source, + filename, + sourceIndex, + platform: meta?.platform || 'unknown', + messageScope: createMessageScope(meta, members, sourceIndex), + messageBridgeFamily: createMessageBridgeFamily(meta), + } + }) + const allMessages: Array<{ + msg: MergerInputMessage + source: string + sourceIndex: number + platform: string + messageScope: string + messageBridgeFamily?: string + }> = [] + + for (const { source, filename, sourceIndex, platform, messageScope, messageBridgeFamily } of preparedSources) { source.streamMessages(10000, (messages) => { for (const msg of messages) { - allMessages.push({ msg, source: filename, platform }) + allMessages.push({ msg, source: filename, sourceIndex, platform, messageScope, messageBridgeFamily }) } }) } - return detectConflictsInMessages(allMessages) + const result = detectConflictsInMessages(allMessages) + const collidingIds = getCollidingPlatformIdsFromMessages(allMessages) + const scopeResolver = createMessageScopeResolver(preparedSources) + for (const message of allMessages) scopeResolver.addMessage(message.msg, message) + scopeResolver.complete() + const normalizeMessageId = createMessageIdNormalizer( + preparedSources.map(({ sourceIndex }) => scopeResolver.resolve(sourceIndex)) + ) + const dedupState = createMessageDedupState() + let totalMessages = 0 + + for (const { msg, sourceIndex, platform } of allMessages) { + const messageScope = scopeResolver.resolve(sourceIndex) + const senderPlatformId = normalizePlatformId(msg.senderPlatformId, platform, collidingIds) + const duplicate = registerMessageAndCheckDuplicate( + { + platformMessageId: normalizeMessageId(msg.platformMessageId, messageScope), + timestamp: msg.timestamp, + senderPlatformId, + type: msg.type, + content: msg.content ?? null, + replyToMessageId: normalizeMessageId(msg.replyToMessageId, messageScope), + }, + dedupState + ) + if (!duplicate) totalMessages++ + } + + return { ...result, totalMessages } } // ==================== Merge orchestration ==================== @@ -108,12 +275,19 @@ export function buildMergedOutput( dataSources: Array<{ source: MergerDataSource; filename: string }>, outputName: string ): MergeOrchestrationResult { - const metas = dataSources.map(({ source, filename }) => ({ - meta: source.getMeta(), - members: source.getMembers(), - filename, - source, - })) + const metas = dataSources.map(({ source, filename }, sourceIndex) => { + const meta = source.getMeta() + const members = source.getMembers() + return { + meta, + members, + filename, + source, + sourceIndex, + messageScope: createMessageScope(meta, members, sourceIndex), + messageBridgeFamily: createMessageBridgeFamily(meta), + } + }) const collidingIds = getCollidingPlatformIds( metas.map(({ meta, members }) => ({ @@ -130,26 +304,71 @@ export function buildMergedOutput( collidingIds ) - // Streaming dedup: process batches to avoid loading all into memory - const seenKeys = new Set() - const mergedMessages: MergedMessage[] = [] + const uniquePlatforms = [...new Set(metas.map(({ meta }) => meta?.platform || 'unknown'))] + const scopeResolver = createMessageScopeResolver(metas) + for (const context of metas) { + if (!context.messageBridgeFamily) continue + context.source.streamMessages(10000, (messages) => { + for (const message of messages) scopeResolver.addMessage(message, context) + }) + } + scopeResolver.complete() + const normalizeMessageId = createMessageIdNormalizer( + metas.map(({ sourceIndex }) => scopeResolver.resolve(sourceIndex)) + ) - for (const { source, meta } of metas) { + // Streaming dedup: prefer stable platform message IDs and use the content + // fingerprint only for messages that do not have one. + const dedupState = createMessageDedupState() + const fallbackOnlyMessageIndexes = new Map() + const mergedMessages: ChatLabMergedMessage[] = [] + + for (const { source, meta, sourceIndex } of metas) { const platform = meta?.platform || 'unknown' + const messageScope = scopeResolver.resolve(sourceIndex) source.streamMessages(10000, (messages) => { for (const msg of messages) { const nid = normalizePlatformId(msg.senderPlatformId, platform, collidingIds) - const key = generateMessageKey(msg.timestamp, nid, msg.content ?? null) - if (seenKeys.has(key)) continue - seenKeys.add(key) + const platformMessageId = normalizeMessageId(msg.platformMessageId, messageScope) + const replyToMessageId = normalizeMessageId(msg.replyToMessageId, messageScope) + const dedupMessage = { + platformMessageId, + timestamp: msg.timestamp, + senderPlatformId: nid, + type: msg.type, + content: msg.content ?? null, + replyToMessageId, + } + const fallbackKey = generateFallbackMessageKey(dedupMessage) + const shouldBackfillId = Boolean( + platformMessageId && + !dedupState.platformMessageIds.has(platformMessageId) && + dedupState.fallbackOnlyKeys.has(fallbackKey) + ) + + if (registerMessageAndCheckDuplicate(dedupMessage, dedupState)) { + if (shouldBackfillId) { + const retainedIndex = fallbackOnlyMessageIndexes.get(fallbackKey) + if (retainedIndex !== undefined) { + mergedMessages[retainedIndex].platformMessageId = platformMessageId + fallbackOnlyMessageIndexes.delete(fallbackKey) + } + } + continue + } + + const retainedIndex = mergedMessages.length mergedMessages.push({ + platformMessageId, sender: nid, accountName: msg.senderAccountName, groupNickname: msg.senderGroupNickname, timestamp: msg.timestamp, type: msg.type, content: msg.content, + replyToMessageId, }) + if (!platformMessageId) fallbackOnlyMessageIndexes.set(fallbackKey, retainedIndex) } }) } @@ -162,7 +381,6 @@ export function buildMergedOutput( messageCount: source.getMessageCount(), })) - const uniquePlatforms = [...new Set(metas.map(({ meta }) => meta?.platform || 'unknown'))] const platform = uniquePlatforms.length === 1 ? uniquePlatforms[0] : 'mixed' const groupIds = new Set(metas.map(({ meta }) => meta?.groupId).filter(Boolean)) diff --git a/packages/node-runtime/src/merger/temp-db.ts b/packages/node-runtime/src/merger/temp-db.ts index 4ead4920..fb817e2e 100644 --- a/packages/node-runtime/src/merger/temp-db.ts +++ b/packages/node-runtime/src/merger/temp-db.ts @@ -12,8 +12,8 @@ import * as fs from 'fs' import * as path from 'path' import type { DatabaseAdapter } from '@openchatlab/core' import { CHATLAB_FORMAT_VERSION, type ParsedMember, type ParsedMessage } from '@openchatlab/shared-types' -import type { MergerDataSource, MergerSourceMeta } from './index' -import type { MergerMember, MergerMessage } from '@openchatlab/core' +import type { MergerDataSource, MergerInputMessage, MergerSourceMeta } from './index' +import type { MergerMember } from '@openchatlab/core' // ==================== Temp DB schema (simplified for merge preview) ==================== @@ -36,12 +36,14 @@ export const TEMP_DB_SCHEMA = ` CREATE TABLE IF NOT EXISTS message ( id INTEGER PRIMARY KEY AUTOINCREMENT, + platform_message_id TEXT, sender_platform_id TEXT NOT NULL, sender_account_name TEXT, sender_group_nickname TEXT, timestamp INTEGER NOT NULL, type INTEGER NOT NULL, - content TEXT + content TEXT, + reply_to_message_id TEXT ); CREATE INDEX IF NOT EXISTS idx_message_ts ON message(timestamp); @@ -88,8 +90,10 @@ export class TempDbWriter { writeMessages(messages: ParsedMessage[]): void { const insert = this.db.prepare( - `INSERT INTO message (sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content) - VALUES (?, ?, ?, ?, ?, ?)` + `INSERT INTO message ( + platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname, + timestamp, type, content, reply_to_message_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` ) const memberInsert = this.db.prepare( 'INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname, avatar) VALUES (?, ?, ?, ?)' @@ -100,12 +104,14 @@ export class TempDbWriter { memberInsert.run(msg.senderPlatformId, msg.senderAccountName || null, msg.senderGroupNickname || null, null) } insert.run( + msg.platformMessageId || null, msg.senderPlatformId, msg.senderAccountName || null, msg.senderGroupNickname || null, msg.timestamp, msg.type, - msg.content || null + msg.content || null, + msg.replyToMessageId || null ) this.messageCount++ } @@ -181,30 +187,35 @@ export class TempDbReader { streamMessages(batchSize: number, callback: (messages: ParsedMessage[]) => void): void { const stmt = this.db.prepare(` - SELECT sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content - FROM message ORDER BY timestamp ASC LIMIT ? OFFSET ? + SELECT platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname, + timestamp, type, content, reply_to_message_id + FROM message ORDER BY timestamp ASC, id ASC LIMIT ? OFFSET ? `) let offset = 0 while (true) { const rows = stmt.all(batchSize, offset) as Array<{ + platform_message_id: string | null sender_platform_id: string sender_account_name: string | null sender_group_nickname: string | null timestamp: number type: number content: string | null + reply_to_message_id: string | null }> if (rows.length === 0) break const messages: ParsedMessage[] = rows.map((r) => ({ + platformMessageId: r.platform_message_id || undefined, senderPlatformId: r.sender_platform_id, senderAccountName: r.sender_account_name || r.sender_platform_id, senderGroupNickname: r.sender_group_nickname || undefined, timestamp: r.timestamp, type: r.type as ParsedMessage['type'], content: r.content, + replyToMessageId: r.reply_to_message_id || undefined, })) callback(messages) @@ -215,25 +226,30 @@ export class TempDbReader { getAllMessages(): ParsedMessage[] { const rows = this.db .prepare( - `SELECT sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content - FROM message ORDER BY timestamp ASC` + `SELECT platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname, + timestamp, type, content, reply_to_message_id + FROM message ORDER BY timestamp ASC, id ASC` ) .all() as Array<{ + platform_message_id: string | null sender_platform_id: string sender_account_name: string | null sender_group_nickname: string | null timestamp: number type: number content: string | null + reply_to_message_id: string | null }> return rows.map((r) => ({ + platformMessageId: r.platform_message_id || undefined, senderPlatformId: r.sender_platform_id, senderAccountName: r.sender_account_name || r.sender_platform_id, senderGroupNickname: r.sender_group_nickname || undefined, timestamp: r.timestamp, type: r.type as ParsedMessage['type'], content: r.content, + replyToMessageId: r.reply_to_message_id || undefined, })) } @@ -271,16 +287,18 @@ function createDataSourceFromReader(reader: TempDbReader): MergerDataSource { getMessageCount(): number { return reader.getMessageCount() }, - streamMessages(batchSize: number, callback: (messages: MergerMessage[]) => void): void { + streamMessages(batchSize: number, callback: (messages: MergerInputMessage[]) => void): void { reader.streamMessages(batchSize, (messages) => { callback( messages.map((msg) => ({ senderPlatformId: msg.senderPlatformId, + platformMessageId: msg.platformMessageId, senderAccountName: msg.senderAccountName, senderGroupNickname: msg.senderGroupNickname, timestamp: msg.timestamp, type: msg.type, content: msg.content, + replyToMessageId: msg.replyToMessageId, })) ) }) @@ -295,12 +313,14 @@ export interface ExportedSession { meta: { name: string; platform: string; type: string; groupId?: string; groupAvatar?: string } members: Array<{ platformId: string; accountName: string; groupNickname?: string; avatar?: string }> messages: Array<{ + platformMessageId?: string sender: string accountName: string groupNickname?: string timestamp: number type: number content: string | null + replyToMessageId?: string }> } @@ -325,16 +345,19 @@ export function exportSessionToJson(db: DatabaseAdapter): ExportedSession { const messages = db .prepare( `SELECT m.platform_id as sender, msg.sender_account_name as accountName, - msg.sender_group_nickname as groupNickname, msg.ts as timestamp, msg.type, msg.content - FROM message msg JOIN member m ON msg.sender_id = m.id ORDER BY msg.ts` + msg.sender_group_nickname as groupNickname, msg.ts as timestamp, msg.type, msg.content, + msg.platform_message_id as platformMessageId, msg.reply_to_message_id as replyToMessageId + FROM message msg JOIN member m ON msg.sender_id = m.id ORDER BY msg.ts, msg.id` ) .all() as Array<{ + platformMessageId?: string sender: string accountName?: string groupNickname?: string timestamp: number type: number content?: string + replyToMessageId?: string }> return { @@ -358,12 +381,14 @@ export function exportSessionToJson(db: DatabaseAdapter): ExportedSession { avatar: m.avatar, })), messages: messages.map((msg) => ({ + platformMessageId: msg.platformMessageId || undefined, sender: msg.sender, accountName: msg.accountName || msg.sender, groupNickname: msg.groupNickname || undefined, timestamp: msg.timestamp, type: msg.type, content: msg.content ?? null, + replyToMessageId: msg.replyToMessageId || undefined, })), } }