watchWithSub<U> method

Stream<List<WithChildren<T, U>>> watchWithSub<U>({ required String subName, required Duration subDefaultExpiration, U subFromJson(Map<String, dynamic>)?, String? subTypeTag, })

Live reactive terminal that joins each parent item matching this query with its children from a named sub-collection.

final stream = todos.query().where((t) => !t.obj.done)
    .watchWithSub<TodoNote>(
      subName: 'notes',
      subDefaultExpiration: const Duration(days: 365),
      subFromJson: TodoNote.fromJson,
      subTypeTag: 'TodoNote',
    );
// stream: Stream<List<WithChildren<Todo, TodoNote>>>

Re-emits on any parent update/delete that could affect the result set, and on any child update/delete within any of the current parents' subName sub-collections. The previous hand-rolled Map<parentId, List<child>> dance in consumer apps collapses to a single stream subscription with this.

The returned stream is single-subscription. Each parent held by the stream owns an internal .watch() on its sub-collection; those child subscriptions are cancelled automatically when the parent leaves the result set (via filter change or delete) and when the outer stream is cancelled.

This is a first-class terminal — implemented here rather than re-invented by every consumer. The children are the sub-collection's full default view.

Implementation

Stream<List<WithChildren<T, U>>> watchWithSub<U>({
  required String subName,
  required Duration subDefaultExpiration,
  U Function(Map<String, dynamic>)? subFromJson,
  String? subTypeTag,
}) {
  // Per-parent state. Key is `<owner>:<id>` — the (owner, id) pair
  // under which a CItem is globally unique.
  final childLatest = <String, List<CItem<U>>>{};
  final childSubs = <String, StreamSubscription<List<CItem<U>>>>{};
  List<CItem<T>> latestParents = const [];
  late final StreamController<List<WithChildren<T, U>>> ctrl;
  StreamSubscription<List<CItem<T>>>? parentSub;

  String keyOf(CItem<T> p) => '${p.owner}:${p.id}';

  void emit() {
    if (ctrl.isClosed) return;
    ctrl.add([
      for (final p in latestParents)
        WithChildren<T, U>(
          parent: p,
          children: List<CItem<U>>.from(childLatest[keyOf(p)] ?? const []),
        ),
    ]);
  }

  Future<void> onParents(List<CItem<T>> parents) async {
    latestParents = parents;
    final currentKeys = parents.map(keyOf).toSet();
    // Cancel subs for parents that left the result set.
    final leavers =
        childSubs.keys.where((k) => !currentKeys.contains(k)).toList();
    for (final k in leavers) {
      await childSubs.remove(k)?.cancel();
      childLatest.remove(k);
    }
    // Track whether we opened any new child sub. If we did, skip
    // the explicit emit below — each new sub's initial-fetch
    // emission already calls emit() through its listener, and
    // emitting from both paths produced duplicate snapshots
    // (consumers that did per-snapshot work, e.g. a TUI sending a
    // read-receipt on every new todo, were doing it twice).
    bool openedNewSub = false;
    for (final p in parents) {
      final k = keyOf(p);
      if (childSubs.containsKey(k)) continue;
      // Use the private internal entry point so we can thread the
      // parent collection's injected notification stream (if any)
      // through to the child sub-collection — required for tests
      // that drive both parent and child events from a single
      // controller. Production callers see only the public
      // [subCollection] verb, which has no notifications: param.
      final sub = _collection._subCollectionInternal<U>(
        parent: p,
        subName: subName,
        defaultExpiration: subDefaultExpiration,
        fromJson: subFromJson,
        typeTag: subTypeTag,
        notifications: _collection._injectedNotifications,
      );
      childSubs[k] = sub.query().watch().listen(
        (children) {
          childLatest[k] = children;
          emit();
        },
        onError: (Object e, StackTrace st) {
          if (!ctrl.isClosed) ctrl.addError(e, st);
        },
      );
      openedNewSub = true;
    }
    // No new subs opened (either only leavers, or the parent set
    // is unchanged) — emit now, otherwise no event is delivered
    // for the leaver removal / parent-set churn.
    if (!openedNewSub) {
      emit();
    }
  }

  ctrl = StreamController<List<WithChildren<T, U>>>(
    onListen: () {
      parentSub = watch().listen(
        (parents) => unawaited(onParents(parents)),
        onError: (Object e, StackTrace st) {
          if (!ctrl.isClosed) ctrl.addError(e, st);
        },
      );
    },
    onCancel: () async {
      await parentSub?.cancel();
      for (final s in childSubs.values) {
        await s.cancel();
      }
      childSubs.clear();
      childLatest.clear();
    },
  );
  return ctrl.stream;
}