fix(postgrest): review fixes for typed filters and asStream

Author: spydon

packages/postgrest/lib/src/postgrest_table.dart

@@ -711,5 +711,14 @@ final class NegatedFilter extends ColumnFilter {
   @override
   PostgrestFilterBuilder<dynamic> _apply(
     PostgrestFilterBuilder<dynamic> builder,
-  ) => builder.not(column, inner.operator, inner.value);
+  ) {
+    // The untyped `not` stringifies map values with `Map.toString`, unlike
+    // the json-encoding positive paths such as `contains`, so encode here.
+    final innerValue = inner.value;
+    return builder.not(
+      column,
+      inner.operator,
+      innerValue is Map ? json.encode(innerValue) : innerValue,
+    );
+  }
 }

packages/postgrest/lib/src/postgrest_typed_builder.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 
 import 'package:postgrest/postgrest.dart';
 
@@ -37,7 +38,24 @@ class PostgrestTypedBuilder<T> implements Future<T> {
   }
 
   @override
-  Stream<T> asStream() => _execute().asStream();
+  Stream<T> asStream() {
+    // Mirrors [PostgrestBuilder.asStream], which returns a broadcast stream.
+    final controller = StreamController<T>.broadcast();
+
+    unawaited(
+      then((value) {
+            controller.add(value);
+          })
+          .catchError((Object error, StackTrace stack) {
+            controller.addError(error, stack);
+          })
+          .whenComplete(() {
+            unawaited(controller.close());
+          }),
+    );
+
+    return controller.stream;
+  }
 
   @override
   Future<T> catchError(Function onError, {bool Function(Object error)? test}) =>

packages/postgrest/lib/src/postgrest_typed_filter_builder.dart

@@ -42,6 +42,13 @@ class PostgrestTypedFilterBuilder<Row, T>
   ///     .whereAny([Books.id.eq(1), Books.title.eq('foo')]);
   /// ```
   PostgrestTypedFilterBuilder<Row, T> whereAny(List<ColumnFilter> filters) {
+    if (filters.isEmpty) {
+      throw ArgumentError.value(
+        filters,
+        'filters',
+        'whereAny needs at least one filter',
+      );
+    }
     final fragments = [for (final filter in filters) _orFragment(filter)];
     return PostgrestTypedFilterBuilder._(
       _filterBuilder.or(fragments.join(',')),
@@ -69,7 +76,9 @@ class PostgrestTypedFilterBuilder<Row, T>
     if (value == null || value is num || value is bool) {
       return '$value';
     }
-    final escaped = '$value'.replaceAll(r'\', r'\\').replaceAll('"', r'\"');
+    // Maps are encoded as json, matching the positive containment paths.
+    final rendered = value is Map ? json.encode(value) : '$value';
+    final escaped = rendered.replaceAll(r'\', r'\\').replaceAll('"', r'\"');
     return '"$escaped"';
   }
 }

packages/postgrest/test/typed_query_test.dart

@@ -18,6 +18,7 @@ class Books {
   static const title = TableColumn<String>('title');
   static const tags = TableColumn<List<String>>('tags');
   static const ageRange = TableColumn<String>('age_range');
+  static const metadata = TableColumn<Map<String, dynamic>>('metadata');
 }
 
 class MockHttpClient extends BaseClient {
@@ -227,6 +228,50 @@ void main() {
     test('negating a filter twice throws', () {
       expect(() => Books.id.eq(1).not().not(), throwsStateError);
     });
+
+    test('negated json containment encodes the value as json', () async {
+      await client
+          .table(Books.table)
+          .select()
+          .where(Books.metadata.contains({'a': 1}).not());
+
+      expect(requestParameters()['metadata'], 'not.cs.{"a":1}');
+    });
+
+    test('whereAny encodes json containment values', () async {
+      await client.table(Books.table).select().whereAny([
+        Books.metadata.contains({'a': 1}),
+        Books.id.eq(1),
+      ]);
+
+      expect(
+        requestParameters()['or'],
+        r'(metadata.cs."{\"a\":1}",id.eq.1)',
+      );
+    });
+
+    test('whereAny without filters throws', () {
+      expect(
+        () => client.table(Books.table).select().whereAny([]),
+        throwsArgumentError,
+      );
+    });
+  });
+
+  group('asStream', () {
+    test('returns a broadcast stream that supports multiple listeners', () {
+      httpClient.responseBody = bookRows;
+
+      final stream = client.table(Books.table).select().asStream();
+
+      expect(stream.isBroadcast, isTrue);
+      stream.listen(
+        expectAsync1((books) {
+          expect(books, hasLength(2));
+        }),
+      );
+      stream.listen(expectAsync1((books) {}));
+    });
   });
 
   group('transforms', () {