Subscribe to source collections and start processing
()
| 441 | |
| 442 | /** Subscribe to source collections and start processing */ |
| 443 | start(): void { |
| 444 | // Use compiled aliases as the source of truth |
| 445 | const compiledAliases = Object.entries(this.compiledAliasToCollectionId) |
| 446 | if (compiledAliases.length === 0) { |
| 447 | // Nothing to subscribe to |
| 448 | return |
| 449 | } |
| 450 | |
| 451 | // When not skipping initial, we always process events immediately |
| 452 | if (!this.skipInitial) { |
| 453 | this.initialLoadComplete = true |
| 454 | } |
| 455 | |
| 456 | // We need to defer initial data processing until ALL subscriptions are |
| 457 | // created, because join pipelines look up subscriptions by alias during |
| 458 | // the graph run. If we run the graph while some aliases are still missing, |
| 459 | // the join tap operator will throw. |
| 460 | // |
| 461 | // Strategy: subscribe to each collection but buffer incoming changes. |
| 462 | // After all subscriptions are in place, flush the buffers and switch to |
| 463 | // direct processing mode. |
| 464 | |
| 465 | const pendingBuffers = new Map< |
| 466 | string, |
| 467 | Array<Array<ChangeMessage<any, string | number>>> |
| 468 | >() |
| 469 | |
| 470 | for (const [alias, collectionId] of compiledAliases) { |
| 471 | const collection = |
| 472 | this.collectionByAlias[alias] ?? this.collections[collectionId]! |
| 473 | |
| 474 | // Initialise per-alias duplicate tracking |
| 475 | this.sentToD2KeysByAlias.set(alias, new Set()) |
| 476 | |
| 477 | // Discover dependencies: if source collection is itself a live query |
| 478 | // collection, its builder must run first during transaction flushes. |
| 479 | const dependencyBuilder = getCollectionBuilder(collection) |
| 480 | if (dependencyBuilder) { |
| 481 | this.aliasDependencies[alias] = [dependencyBuilder] |
| 482 | this.builderDependencies.add(dependencyBuilder) |
| 483 | } else { |
| 484 | this.aliasDependencies[alias] = [] |
| 485 | } |
| 486 | |
| 487 | // Get where clause for this alias (for predicate push-down) |
| 488 | const whereClause = this.sourceWhereClauses?.get(alias) |
| 489 | const whereExpression = whereClause |
| 490 | ? normalizeExpressionPaths(whereClause, alias) |
| 491 | : undefined |
| 492 | |
| 493 | // Initialise buffer for this alias |
| 494 | const buffer: Array<Array<ChangeMessage<any, string | number>>> = [] |
| 495 | pendingBuffers.set(alias, buffer) |
| 496 | |
| 497 | // Lazy aliases (marked by the join compiler) should NOT load initial state |
| 498 | // eagerly — the join tap operator will load exactly the rows it needs on demand. |
| 499 | // For on-demand collections, eager loading would trigger a full server fetch |
| 500 | // for data that should be lazily loaded based on join keys. |
no test coverage detected