watchAll method

Stream<List<T>> watchAll({ String? userId, bool includeInitialData = true, List<String> withRelated = const [], bool includeDeleted = false, })

Watches all entities from the local adapter, emitting a new list on any change.

The includeInitialData parameter controls whether the stream should immediately emit the current list of all items. Defaults to true. If false, the stream will only emit when a change occurs. Returns null if the adapter does not support reactive queries.

Implementation

Stream<List<T>> watchAll({String? userId, bool includeInitialData = true, List<String> withRelated = const [], bool includeDeleted = false}) {
  _ensureInitialized();
  final adapterStream = localAdapter.watchAll(userId: userId, includeInitialData: includeInitialData);
  if (adapterStream == null) {
    _logger.debug('${localAdapter.name} is not watchable; watchAll returns an empty stream. Mix in WatchableAdapter for reactivity.');
    return Stream<List<T>>.empty();
  }

  return adapterStream.asyncMap((emitted) async {
    final list = includeDeleted ? emitted : emitted.where((e) => !e.isDeleted).toList();
    try {
      // Eagerly load requested relations on every emission (reactive #UX).
      if (withRelated.isNotEmpty && list.isNotEmpty) {
        await _fetchAndStitchRelations(list, withRelated, DataSource.local, userId);
      }
      // Apply post-fetch transforms with error handling
      final transformedList = <T>[];
      for (final entity in list) {
        try {
          final transformed = await _applyPostFetchTransforms(entity);
          transformedList.add(transformed);
        } catch (e, stack) {
          _logger.error('Failed to apply post-fetch transforms to entity ${entity.id}: $e', stack);
          // Continue with other entities instead of failing the entire stream
          transformedList.add(entity); // Use original entity if transform fails
        }
      }
      return transformedList;
    } catch (e, stack) {
      _logger.error('Failed to transform entity list in watchAll: $e', stack);
      // Return the original list if transformation fails completely
      return list;
    }
  }).handleError((error, stack) {
    _logger.error('Error in watchAll stream for $T: $error', stack);
    // Don't rethrow - let the stream continue
  });
}