Function setupNestedPipelines

packages/db/src/query/live/collection-config-builder.ts:1284–1352

* Sets up shared buffers for nested includes pipelines. * Instead of writing directly into a single shared IncludesOutputState, * each nested pipeline writes into a buffer that is later drained per-entry.

(
  includes: Array<IncludesCompilationResult>,
  syncState: SyncState,
)

 * each nested pipeline writes into a buffer that is later drained per-entry.
 */
function setupNestedPipelines(
  includes: Array<IncludesCompilationResult>,
  syncState: SyncState,
): Array<NestedIncludesSetup> {
  return includes.map((entry) => {
    const buffer: Map<unknown, Map<unknown, Changes<any>>> = new Map()

    // Attach output callback that writes into the shared buffer
    entry.pipeline.pipe(
      output((data) => {
        const messages = data.getInner()
        syncState.messagesCount += messages.length

        for (const [[childKey, tupleData], multiplicity] of messages) {
          const [childResult, _orderByIndex, correlationKey, parentContext] =
            tupleData as unknown as [
              any,
              string | undefined,
              unknown,
              Record<string, any> | null,
            ]

          const routingKey = computeRoutingKey(correlationKey, parentContext)

          let byChild = buffer.get(routingKey)
          if (!byChild) {
            byChild = new Map()
            buffer.set(routingKey, byChild)
          }

          const existing = byChild.get(childKey) || {
            deletes: 0,
            inserts: 0,
            value: childResult,
            orderByIndex: _orderByIndex,
          }

          if (multiplicity < 0) {
            existing.deletes += Math.abs(multiplicity)
          } else if (multiplicity > 0) {
            existing.inserts += multiplicity
            existing.value = childResult
          }

          byChild.set(childKey, existing)
        }
      }),
    )

    const setup: NestedIncludesSetup = {
      compilationResult: entry,
      buffer,
      snapshot: new Map(),
      routingIndex: new Map(),
      routingReverseIndex: new Map(),
      routingChildToNested: new Map(),
    }