diff --git a/apps/cli/src/import/stream-import-flow.test.ts b/apps/cli/src/import/stream-import-flow.test.ts index 6510de60..dd7f8d52 100644 --- a/apps/cli/src/import/stream-import-flow.test.ts +++ b/apps/cli/src/import/stream-import-flow.test.ts @@ -67,6 +67,27 @@ function writeTempDuplicateChatFile(dir: string): string { return filePath } +function writeWeFlowImageExport(dir: string, filename: string, messageCount: number): string { + const filePath = path.join(dir, filename) + fs.writeFileSync( + filePath, + JSON.stringify({ + weflow: { version: '1.0.0' }, + session: { wxid: 'room@chatroom', displayName: 'Image Room', type: '群聊' }, + messages: Array.from({ length: messageCount }, (_, index) => ({ + localId: index + 1, + createTime: 1711468800, + type: '图片消息', + content: '[图片]', + isSend: 0, + senderUsername: 'wxid_alice', + senderDisplayName: 'Alice', + })), + }) + ) + return filePath +} + function writeBatchChatFile(dir: string, filename: string, groupId: string, content: string): string { const filePath = path.join(dir, filename) fs.writeFileSync( @@ -475,6 +496,37 @@ test('autoImport reports the same exact-message deduplication on create and incr assert.equal(row.count, 1) }) +test('autoImport preserves same-second WeFlow message counts across overlapping exports', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) + const manager = new DatabaseManager(createPathProvider(root), { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + const firstExport = writeWeFlowImageExport(root, 'weflow-first.json', 2) + const overlappingExport = writeWeFlowImageExport(root, 'weflow-overlap.json', 3) + + const created = await autoImport(manager, firstExport, { nativeBinding }) + const incremental = await autoImport(manager, overlappingExport, { nativeBinding }) + const repeated = await autoImport(manager, overlappingExport, { nativeBinding }) + + assert.equal(created.success, true) + assert.equal(created.newMessageCount, 2) + assert.equal(created.duplicateCount, 0) + assert.equal(incremental.success, true) + assert.equal(incremental.newMessageCount, 1) + assert.equal(incremental.duplicateCount, 2) + assert.equal(repeated.success, true) + assert.equal(repeated.newMessageCount, 0) + assert.equal(repeated.duplicateCount, 3) + + const db = manager.openRawSessionDatabase(created.sessionId!, { readonly: true }) + const row = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number } + db.close() + assert.equal(row.count, 3) +}) + test('autoImport uses an explicit id for create and then forces incremental import', async () => { const root = makeTempDir() fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) diff --git a/packages/node-runtime/src/import/incremental-importer.test.ts b/packages/node-runtime/src/import/incremental-importer.test.ts index a18355e9..1a6847f9 100644 --- a/packages/node-runtime/src/import/incremental-importer.test.ts +++ b/packages/node-runtime/src/import/incremental-importer.test.ts @@ -512,6 +512,60 @@ test('preserves canonical dedup semantics while checking existing database candi } }) +test('bridges stable IDs to repeated fallback-only occurrences one for one', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const dbPath = path.join(tempDir, 'session.db') + const filePath = path.join(tempDir, 'incoming.json') + seedSessionDb(dbPath) + seedExistingMessage(dbPath, {}) + + const seededDb = openBetterSqliteDatabase(dbPath, { nativeBinding }) + const member = seededDb.prepare('SELECT id FROM member WHERE platform_id = ?').get('wxid_alice') as { id: number } + seededDb + .prepare('INSERT INTO message (sender_id, ts, type, content) VALUES (?, ?, ?, ?)') + .run(member.id, 1780330832, 0, 'same message') + seededDb.close() + + fs.writeFileSync( + filePath, + JSON.stringify({ + weflow: { version: '1.0.0' }, + session: { wxid: 'wxid_alice', displayName: 'Alice', type: '私聊' }, + messages: ['msg-1', 'msg-2', 'msg-3'].map((platformMessageId, index) => ({ + localId: index + 1, + platformMessageId, + createTime: 1780330832, + type: '文本消息', + content: 'same message', + senderUsername: 'wxid_alice', + senderDisplayName: 'Alice', + })), + }) + ) + + const deps = createDeps(dbPath) + const analysis = await analyzeIncrementalImport('session', filePath, deps, { formatId: 'weflow' }) + assert.equal(analysis.newMessageCount, 1) + assert.equal(analysis.duplicateCount, 2) + + const result = await incrementalImport('session', filePath, deps, { formatId: 'weflow' }) + assert.equal(result.success, true) + assert.equal(result.newMessageCount, 1) + assert.equal(result.batch?.duplicateCount, 2) + + const db = openBetterSqliteDatabase(dbPath, { readonly: true, nativeBinding }) + const rows = db.prepare('SELECT platform_message_id FROM message ORDER BY id').all() as Array<{ + platform_message_id: string | null + }> + db.close() + assert.deepEqual( + rows.map((row) => row.platform_message_id), + [null, null, 'msg-3'] + ) +}) + test('deduplicates repeated input across parser message batches', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index 3c70894e..860734b4 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -25,7 +25,12 @@ import { type DedupMessage, type MessageDedupState, } from './message-deduplicator' -import { normalizeSystemMemberName, SYSTEM_MEMBER_NAME, type ImportProgressCallback } from './streaming-importer' +import { + normalizeSystemMemberName, + shouldPreserveFallbackMultiplicity, + SYSTEM_MEMBER_NAME, + type ImportProgressCallback, +} from './streaming-importer' // ==================== Public interfaces ==================== @@ -109,8 +114,9 @@ interface ExistingMessageLookup { db: DatabaseAdapter maxMessageId: number hasFallbackOnlyMessages: boolean - consumedFallbackOnlyKeys: Set + consumedFallbackOnlyOccurrenceCounts: Map bridgedPlatformMessageIds: Set + consumedFallbackOccurrenceCounts: Map candidateStatement: PreparedStatement platformIdStatements: Map } @@ -159,8 +165,9 @@ function createExistingMessageLookup(db: DatabaseAdapter): ExistingMessageLookup db, maxMessageId, hasFallbackOnlyMessages, - consumedFallbackOnlyKeys: new Set(), + consumedFallbackOnlyOccurrenceCounts: new Map(), bridgedPlatformMessageIds: new Set(), + consumedFallbackOccurrenceCounts: new Map(), candidateStatement: db.prepare( `SELECT msg.type, msg.content, msg.reply_to_message_id, msg.platform_message_id FROM member m @@ -229,7 +236,8 @@ function isExistingMessageDuplicate( lookup: ExistingMessageLookup, message: DedupMessage, existingBatchPlatformIds: Set, - candidateCache: CandidateCache + candidateCache: CandidateCache, + preserveFallbackMultiplicity: boolean ): boolean { if ( message.platformMessageId && @@ -241,20 +249,30 @@ function isExistingMessageDuplicate( const fallbackKey = generateFallbackMessageKey(message) if (message.platformMessageId) { - if (!lookup.hasFallbackOnlyMessages || lookup.consumedFallbackOnlyKeys.has(fallbackKey)) return false - const matchesFallbackOnly = getExistingCandidates(lookup, message, candidateCache).some( + if (!lookup.hasFallbackOnlyMessages) return false + const matchingFallbackOnlyCount = getExistingCandidates(lookup, message, candidateCache).filter( (candidate) => !candidate.platform_message_id && candidateFallbackKey(message, candidate) === fallbackKey - ) + ).length + const consumedCount = lookup.consumedFallbackOnlyOccurrenceCounts.get(fallbackKey) ?? 0 + const matchesFallbackOnly = preserveFallbackMultiplicity + ? consumedCount < matchingFallbackOnlyCount + : consumedCount === 0 && matchingFallbackOnlyCount > 0 if (matchesFallbackOnly) { - lookup.consumedFallbackOnlyKeys.add(fallbackKey) + lookup.consumedFallbackOnlyOccurrenceCounts.set(fallbackKey, consumedCount + 1) lookup.bridgedPlatformMessageIds.add(message.platformMessageId) } return matchesFallbackOnly } - return getExistingCandidates(lookup, message, candidateCache).some( + const matchingCount = getExistingCandidates(lookup, message, candidateCache).filter( (candidate) => candidateFallbackKey(message, candidate) === fallbackKey - ) + ).length + if (!preserveFallbackMultiplicity) return matchingCount > 0 + + const consumedCount = lookup.consumedFallbackOccurrenceCounts.get(fallbackKey) ?? 0 + if (consumedCount >= matchingCount) return false + lookup.consumedFallbackOccurrenceCounts.set(fallbackKey, consumedCount + 1) + return true } function isDuplicate( @@ -262,9 +280,13 @@ function isDuplicate( incomingState: MessageDedupState, message: DedupMessage, existingBatchPlatformIds: Set, - candidateCache: CandidateCache + candidateCache: CandidateCache, + preserveFallbackMultiplicity: boolean ): boolean { - if (isExistingMessageDuplicate(lookup, message, existingBatchPlatformIds, candidateCache)) return true + if ( + isExistingMessageDuplicate(lookup, message, existingBatchPlatformIds, candidateCache, preserveFallbackMultiplicity) + ) + return true return registerMessageAndCheckDuplicate(message, incomingState) } @@ -298,10 +320,11 @@ export async function analyzeIncrementalImport( let duplicateCount = 0 let platform = formatFeature.platform const mapPlatformId = createPlatformIdMapper(options?.senderPlatformIdMappings) + const preserveFallbackMultiplicity = shouldPreserveFallbackMultiplicity(formatFeature.id) try { const existingLookup = createExistingMessageLookup(db) - const incomingState = createMessageDedupState() + const incomingState = createMessageDedupState([], [], [], { preserveFallbackMultiplicity }) await streamParseFile( filePath, @@ -336,7 +359,8 @@ export async function analyzeIncrementalImport( incomingState, { ...msg, timestamp }, existingBatchPlatformIds, - candidateCache + candidateCache, + preserveFallbackMultiplicity ) ) { duplicateCount++ @@ -378,10 +402,11 @@ export async function incrementalImport( const metaUpdateMode = options?.metaUpdateMode ?? 'patch' const memberUpdateMode = options?.memberUpdateMode ?? 'upsert' const mapPlatformId = createPlatformIdMapper(options?.senderPlatformIdMappings) + const preserveFallbackMultiplicity = shouldPreserveFallbackMultiplicity(formatFeature.id) try { const existingLookup = createExistingMessageLookup(db) - const incomingState = createMessageDedupState() + const incomingState = createMessageDedupState([], [], [], { preserveFallbackMultiplicity }) const memberIdMap = new Map() const existingMembers = db.prepare('SELECT id, platform_id FROM member').all() as Array<{ @@ -526,7 +551,8 @@ export async function incrementalImport( incomingState, { ...msg, timestamp }, existingBatchPlatformIds, - candidateCache + candidateCache, + preserveFallbackMultiplicity ) ) { duplicateCount++ diff --git a/packages/node-runtime/src/import/message-deduplicator.test.ts b/packages/node-runtime/src/import/message-deduplicator.test.ts index c26e28f9..2a0f829d 100644 --- a/packages/node-runtime/src/import/message-deduplicator.test.ts +++ b/packages/node-runtime/src/import/message-deduplicator.test.ts @@ -33,3 +33,26 @@ test('bridges one fallback-only copy without collapsing later distinct platform assert.equal(registerMessageAndCheckDuplicate({ ...baseMessage, platformMessageId: 'msg-1' }, state), true) assert.equal(registerMessageAndCheckDuplicate({ ...baseMessage, platformMessageId: 'msg-2' }, state), false) }) + +test('preserves repeated fallback occurrences when multiplicity is enabled', () => { + const state = createMessageDedupState([], [], [], { preserveFallbackMultiplicity: true }) + + assert.equal(registerMessageAndCheckDuplicate(baseMessage, state), false) + assert.equal(registerMessageAndCheckDuplicate(baseMessage, state), false) +}) + +test('bridges fallback-only and stable-ID occurrences one for one', () => { + const fallbackFirst = createMessageDedupState([], [], [], { preserveFallbackMultiplicity: true }) + assert.equal(registerMessageAndCheckDuplicate(baseMessage, fallbackFirst), false) + assert.equal(registerMessageAndCheckDuplicate(baseMessage, fallbackFirst), false) + assert.equal(registerMessageAndCheckDuplicate({ ...baseMessage, platformMessageId: 'msg-1' }, fallbackFirst), true) + assert.equal(registerMessageAndCheckDuplicate({ ...baseMessage, platformMessageId: 'msg-2' }, fallbackFirst), true) + assert.equal(registerMessageAndCheckDuplicate({ ...baseMessage, platformMessageId: 'msg-3' }, fallbackFirst), false) + + const stableFirst = createMessageDedupState([], [], [], { preserveFallbackMultiplicity: true }) + assert.equal(registerMessageAndCheckDuplicate({ ...baseMessage, platformMessageId: 'msg-1' }, stableFirst), false) + assert.equal(registerMessageAndCheckDuplicate({ ...baseMessage, platformMessageId: 'msg-2' }, stableFirst), false) + assert.equal(registerMessageAndCheckDuplicate(baseMessage, stableFirst), true) + assert.equal(registerMessageAndCheckDuplicate(baseMessage, stableFirst), true) + assert.equal(registerMessageAndCheckDuplicate(baseMessage, stableFirst), false) +}) diff --git a/packages/node-runtime/src/import/message-deduplicator.ts b/packages/node-runtime/src/import/message-deduplicator.ts index 1042e96d..7da16206 100644 --- a/packages/node-runtime/src/import/message-deduplicator.ts +++ b/packages/node-runtime/src/import/message-deduplicator.ts @@ -8,6 +8,14 @@ export interface MessageDedupState { fallbackKeys: Set /** Unmatched fallback keys whose accepted message did not have a stable platform ID. */ fallbackOnlyKeys: Set + preserveFallbackMultiplicity: boolean + /** Positive = unmatched ID-less occurrences; negative = unmatched stable-ID occurrences. */ + fallbackMultiplicityBalances: Map +} + +export interface MessageDedupOptions { + /** Preserve repeated ID-less messages that share a fallback fingerprint. */ + preserveFallbackMultiplicity?: boolean } export interface DedupMessage { @@ -48,15 +56,36 @@ export function generateFallbackMessageKey(message: Omit = [], fallbackKeys: Iterable = [], - fallbackOnlyKeys: Iterable = [] + fallbackOnlyKeys: Iterable = [], + options: MessageDedupOptions = {} ): MessageDedupState { + const fallbackKeySet = new Set(fallbackKeys) + const fallbackOnlyKeySet = new Set(fallbackOnlyKeys) + const fallbackMultiplicityBalances = new Map() + if (options.preserveFallbackMultiplicity) { + for (const key of fallbackKeySet) fallbackMultiplicityBalances.set(key, -1) + for (const key of fallbackOnlyKeySet) fallbackMultiplicityBalances.set(key, 1) + } return { platformMessageIds: new Set(platformMessageIds), - fallbackKeys: new Set(fallbackKeys), - fallbackOnlyKeys: new Set(fallbackOnlyKeys), + fallbackKeys: fallbackKeySet, + fallbackOnlyKeys: fallbackOnlyKeySet, + preserveFallbackMultiplicity: options.preserveFallbackMultiplicity ?? false, + fallbackMultiplicityBalances, } } +function setFallbackMultiplicityBalance(state: MessageDedupState, key: string, balance: number): void { + if (balance === 0) state.fallbackMultiplicityBalances.delete(key) + else state.fallbackMultiplicityBalances.set(key, balance) +} + +export function hasUnmatchedFallbackOnlyMessage(state: MessageDedupState, fallbackKey: string): boolean { + return state.preserveFallbackMultiplicity + ? (state.fallbackMultiplicityBalances.get(fallbackKey) ?? 0) > 0 + : state.fallbackOnlyKeys.has(fallbackKey) +} + /** * Canonical import dedup rule shared by file imports and JSON Push: prefer stable platform * message IDs and fall back to timestamp + sender platform ID + type + normalized content + reply target. @@ -68,6 +97,16 @@ export function registerMessageAndCheckDuplicate(message: DedupMessage, state: M if (state.platformMessageIds.has(message.platformMessageId)) return true state.platformMessageIds.add(message.platformMessageId) + if (state.preserveFallbackMultiplicity) { + const balance = state.fallbackMultiplicityBalances.get(key) ?? 0 + if (balance > 0) { + setFallbackMultiplicityBalance(state, key, balance - 1) + return true + } + setFallbackMultiplicityBalance(state, key, balance - 1) + return false + } + // Only bridge ID-bearing messages to fallback-only copies. Comparing this // key against other ID-bearing messages would incorrectly merge distinct // platform messages that happen to have identical content and timestamps. @@ -76,6 +115,16 @@ export function registerMessageAndCheckDuplicate(message: DedupMessage, state: M return false } + if (state.preserveFallbackMultiplicity) { + const balance = state.fallbackMultiplicityBalances.get(key) ?? 0 + if (balance < 0) { + setFallbackMultiplicityBalance(state, key, balance + 1) + return true + } + setFallbackMultiplicityBalance(state, key, balance + 1) + return false + } + if (state.fallbackKeys.has(key)) return true state.fallbackKeys.add(key) state.fallbackOnlyKeys.add(key) diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index aed6cad1..44515755 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -24,6 +24,7 @@ import { getPreprocessor, needsPreprocess, isNativeFormatAvailable, + PARSER_FORMAT_IDS, type ParsedMeta, type FormatFeature, type ParseProgress, @@ -33,6 +34,10 @@ import { performance } from 'node:perf_hooks' import { MessageBatchInserter, type MessageInsertRow } from './message-batch-inserter' import { createMessageDedupState, registerMessageAndCheckDuplicate, type DedupMessage } from './message-deduplicator' +export function shouldPreserveFallbackMultiplicity(formatId: string): boolean { + return formatId === PARSER_FORMAT_IDS.WEFLOW || formatId === PARSER_FORMAT_IDS.ECHOTRACE +} + // ==================== Public interfaces ==================== export interface SkipReasons { @@ -524,7 +529,9 @@ async function streamImportSingle( skippedInvalidTimestamp: 0, skippedNoType: 0, } - const dedupState = createMessageDedupState() + const dedupState = createMessageDedupState([], [], [], { + preserveFallbackMultiplicity: shouldPreserveFallbackMultiplicity(formatFeature.id), + }) logger?.info('Starting streamParseFile...') const parsePipelineStartedAt = now() @@ -940,7 +947,9 @@ export async function analyzeNewImport( let meta: { name: string; platform: string; type: string } | null = null const memberSet = new Set() - const dedupState = createMessageDedupState() + const dedupState = createMessageDedupState([], [], [], { + preserveFallbackMultiplicity: shouldPreserveFallbackMultiplicity(formatFeature.id), + }) let totalMessages = 0 let newMessageCount = 0 let duplicateCount = 0 diff --git a/packages/node-runtime/src/merger/__tests__/merger.test.ts b/packages/node-runtime/src/merger/__tests__/merger.test.ts index 4dd0e9e2..a6a4df3b 100644 --- a/packages/node-runtime/src/merger/__tests__/merger.test.ts +++ b/packages/node-runtime/src/merger/__tests__/merger.test.ts @@ -111,6 +111,38 @@ describe('buildMergedOutput', () => { assert.equal(result.chatLabData.meta.sources.length, 2) }) + it('preserves the maximum fallback occurrence count across overlapping sources', () => { + const meta: MergerSourceMeta = { name: 'chat', platform: 'wechat', type: 'group', groupId: 'room' } + const repeatedImage: MergerInputMessage = { + senderPlatformId: 'u1', + timestamp: 100, + type: 1, + content: '[图片]', + } + const source1 = createMockSource( + meta, + [{ platformId: 'u1', accountName: 'Alice' }], + [{ ...repeatedImage }, { ...repeatedImage }] + ) + const source2 = createMockSource( + meta, + [{ platformId: 'u1', accountName: 'Alice' }], + [{ ...repeatedImage }, { ...repeatedImage }, { ...repeatedImage }] + ) + + const sources = [ + { source: source1, filename: 'first.json' }, + { source: source2, filename: 'overlap.json' }, + ] + const result = buildMergedOutput(sources, 'Merged') + const reversed = buildMergedOutput([...sources].reverse(), 'Merged') + const conflicts = checkConflictsFromSources(sources) + + assert.equal(result.chatLabData.messages.length, 3) + assert.equal(reversed.chatLabData.messages.length, 3) + assert.equal(conflicts.totalMessages, 3) + }) + it('detects mixed platform when sources differ', () => { const s1 = createMockSource( { name: 'a', platform: 'qq', type: 'group' }, diff --git a/packages/node-runtime/src/merger/index.ts b/packages/node-runtime/src/merger/index.ts index 490463a6..9a93538e 100644 --- a/packages/node-runtime/src/merger/index.ts +++ b/packages/node-runtime/src/merger/index.ts @@ -21,6 +21,7 @@ import { CHATLAB_FORMAT_VERSION } from '@openchatlab/shared-types' import { createMessageDedupState, generateFallbackMessageKey, + hasUnmatchedFallbackOnlyMessage, registerMessageAndCheckDuplicate, } from '../import/message-deduplicator' @@ -95,6 +96,37 @@ interface MessageScopeContext { messageBridgeFamily?: string } +function createFallbackOccurrenceTracker(): (message: MergerInputMessage, sourceIndex: number) => boolean { + // Overlapping exports contribute the maximum occurrence count for each + // fallback key, rather than collapsing to one message or summing all copies. + const maximumOccurrences = new Map() + const sourceOccurrences = new Map>() + + return (message, sourceIndex) => { + if (message.platformMessageId) return false + + const fallbackKey = generateFallbackMessageKey({ + timestamp: message.timestamp, + senderPlatformId: message.senderPlatformId, + type: message.type, + content: message.content ?? null, + replyToMessageId: message.replyToMessageId, + }) + let occurrences = sourceOccurrences.get(sourceIndex) + if (!occurrences) { + occurrences = new Map() + sourceOccurrences.set(sourceIndex, occurrences) + } + const occurrence = (occurrences.get(fallbackKey) ?? 0) + 1 + occurrences.set(fallbackKey, occurrence) + + const previousMaximum = maximumOccurrences.get(fallbackKey) ?? 0 + if (occurrence <= previousMaximum) return true + maximumOccurrences.set(fallbackKey, occurrence) + return false + } +} + function createMessageScopeResolver(contexts: Iterable): { addMessage: (message: MergerInputMessage, context: MessageScopeContext) => void complete: () => void @@ -241,23 +273,24 @@ export function checkConflictsFromSources( const normalizeMessageId = createMessageIdNormalizer( preparedSources.map(({ sourceIndex }) => scopeResolver.resolve(sourceIndex)) ) - const dedupState = createMessageDedupState() + const dedupState = createMessageDedupState([], [], [], { preserveFallbackMultiplicity: true }) + const isCoveredFallbackOccurrence = createFallbackOccurrenceTracker() 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 - ) + const dedupMessage = { + platformMessageId: normalizeMessageId(msg.platformMessageId, messageScope), + timestamp: msg.timestamp, + senderPlatformId, + type: msg.type, + content: msg.content ?? null, + replyToMessageId: normalizeMessageId(msg.replyToMessageId, messageScope), + } + const duplicate = + isCoveredFallbackOccurrence(dedupMessage, sourceIndex) || + registerMessageAndCheckDuplicate(dedupMessage, dedupState) if (!duplicate) totalMessages++ } @@ -324,8 +357,9 @@ export function buildMergedOutput( // 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 dedupState = createMessageDedupState([], [], [], { preserveFallbackMultiplicity: true }) + const isCoveredFallbackOccurrence = createFallbackOccurrenceTracker() + const fallbackOnlyMessageIndexes = new Map() const mergedMessages: ChatLabMergedMessage[] = [] for (const { source, meta, sourceIndex } of metas) { @@ -345,18 +379,20 @@ export function buildMergedOutput( replyToMessageId, } const fallbackKey = generateFallbackMessageKey(dedupMessage) + if (isCoveredFallbackOccurrence(dedupMessage, sourceIndex)) continue const shouldBackfillId = Boolean( platformMessageId && !dedupState.platformMessageIds.has(platformMessageId) && - dedupState.fallbackOnlyKeys.has(fallbackKey) + hasUnmatchedFallbackOnlyMessage(dedupState, fallbackKey) ) if (registerMessageAndCheckDuplicate(dedupMessage, dedupState)) { if (shouldBackfillId) { - const retainedIndex = fallbackOnlyMessageIndexes.get(fallbackKey) + const retainedIndexes = fallbackOnlyMessageIndexes.get(fallbackKey) + const retainedIndex = retainedIndexes?.shift() if (retainedIndex !== undefined) { mergedMessages[retainedIndex].platformMessageId = platformMessageId - fallbackOnlyMessageIndexes.delete(fallbackKey) + if (retainedIndexes?.length === 0) fallbackOnlyMessageIndexes.delete(fallbackKey) } } continue @@ -373,7 +409,11 @@ export function buildMergedOutput( content: msg.content, replyToMessageId, }) - if (!platformMessageId) fallbackOnlyMessageIndexes.set(fallbackKey, retainedIndex) + if (!platformMessageId) { + const retainedIndexes = fallbackOnlyMessageIndexes.get(fallbackKey) ?? [] + retainedIndexes.push(retainedIndex) + fallbackOnlyMessageIndexes.set(fallbackKey, retainedIndexes) + } } }) }