Function buildMergedOutput

packages/node-runtime/src/merger/index.ts:106–193

(
  dataSources: Array<{ source: MergerDataSource; filename: string }>,
  outputName: string
)

 * assembles the output. Does NOT write to disk or import to DB.
 */
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 collidingIds = getCollidingPlatformIds(
    metas.map(({ meta, members }) => ({
      platform: meta?.platform || 'unknown',
      members: members.map((m) => ({ platformId: m.platformId })),
    }))
  )

  const memberMap = mergeMergerMembers(
    metas.map(({ meta, members }) => ({
      platform: meta?.platform || 'unknown',
      members,
    })),
    collidingIds
  )

  // Streaming dedup: process batches to avoid loading all into memory
  const seenKeys = new Set<string>()
  const mergedMessages: MergedMessage[] = []

  for (const { source, meta } of metas) {
    const platform = meta?.platform || 'unknown'
    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)
        mergedMessages.push({
          sender: nid,
          accountName: msg.senderAccountName,
          groupNickname: msg.senderGroupNickname,
          timestamp: msg.timestamp,
          type: msg.type,
          content: msg.content,
        })
      }
    })
  }

  mergedMessages.sort((a, b) => a.timestamp - b.timestamp)

  const sources: MergeSourceInfo[] = dataSources.map(({ source, filename }) => ({
    filename,
    platform: source.getMeta()?.platform || 'unknown',
    messageCount: source.getMessageCount(),
  }))