Firestore/pipelines

packages/firebase_admin_sdk/example/lib/firestore_example.dart

@@ -33,6 +33,7 @@ Future<void> firestoreExample(FirebaseApp admin) async {
   await recursiveDeleteExample(admin);
   await bulkWriterExamples(admin);
   await bundleBuilderExample(admin);
+  await pipelineExample(admin);
 }
 
 /// Example 1: Basic Firestore operations with default database
@@ -507,3 +508,74 @@ Future<void> recursiveDeleteExample(FirebaseApp admin) async {
     print('> Error: $e');
   }
 }
+
+/// Pipeline example demonstrating the Firestore Pipelines API.
+///
+/// Pipelines offer a composable, stage-based alternative to structured queries.
+/// Each stage transforms the stream of documents: filter, project, sort,
+/// aggregate, and more — all in a single server-side execution.
+Future<void> pipelineExample(FirebaseApp admin) async {
+  print('### Pipeline Example ###\n');
+
+  final firestore = admin.firestore(databaseId: 'dart-admin-enterprise');
+  final col = firestore.collection('pipeline-demo');
+
+  // Seed some product data.
+  await Future.wait([
+    col.doc('p1').set({'name': 'Widget', 'category': 'hardware', 'price': 9.99, 'stock': 100}),
+    col.doc('p2').set({'name': 'Gadget', 'category': 'electronics', 'price': 49.99, 'stock': 25}),
+    col.doc('p3').set({'name': 'Doohickey', 'category': 'hardware', 'price': 4.99, 'stock': 200}),
+    col.doc('p4').set({'name': 'Thingamajig', 'category': 'electronics', 'price': 99.99, 'stock': 10}),
+    col.doc('p5').set({'name': 'Whatsit', 'category': 'hardware', 'price': 14.99, 'stock': 75}),
+  ]);
+
+  try {
+    // --- Stage 1: filter → select → sort ---
+    print('> Hardware products (name + price, cheapest first):\n');
+
+    final filtered = await firestore
+        .pipeline()
+        .collection('pipeline-demo')
+        .where(
+          Expression.field('category').equal(Expression.constant('hardware')),
+        )
+        .select(['name', 'price'])
+        .sort([Ordering.ascending(Expression.field('price'))])
+        .execute();
+
+    for (final result in filtered.results) {
+      print('  - ${result.data()['name']}: \$${result.data()['price']}');
+    }
+    print('');
+
+    // --- Stage 2: aggregate — count + average price per category ---
+    print('> Count and average price per category:\n');
+
+    final aggregated = await firestore
+        .pipeline()
+        .collection('pipeline-demo')
+        .aggregate(
+          accumulators: [
+            AggregateFunction.countAll().as('count'),
+            AggregateFunction.average('price').as('avg_price'),
+          ],
+          groupBy: [Expression.field('category')],
+        )
+        .sort([Ordering.ascending(Expression.field('category'))])
+        .execute();
+
+    for (final result in aggregated.results) {
+      final d = result.data();
+      final avg = (d['avg_price'] as num?)?.toStringAsFixed(2) ?? 'n/a';
+      print('  - ${d['category']}: ${d['count']} product(s), avg \$$avg');
+    }
+    print('');
+  } catch (e) {
+    print('> Error: $e');
+  } finally {
+    // Clean up seeded data.
+    await Future.wait([
+      for (final id in ['p1', 'p2', 'p3', 'p4', 'p5']) col.doc(id).delete(),
+    ]);
+  }
+}

packages/firebase_admin_sdk/test/integration/app/firebase_app_prod_test.dart

@@ -44,19 +44,23 @@ void main() {
         timeout: const Timeout(Duration(seconds: 30)),
       );
 
-      test('SDK-created ADC client is closed when app.close() is called', () {
-        return runZoned(() async {
-          final app = FirebaseApp.initializeApp(
-            name: 'adc-close-${DateTime.now().microsecondsSinceEpoch}',
-            options: const AppOptions(projectId: projectId),
-          );
-
-          await app.client;
-          await app.close();
-
-          expect(app.isDeleted, isTrue);
-        }, zoneValues: {envSymbol: prodEnv()});
-      }, timeout: const Timeout(Duration(seconds: 30)));
+      test(
+        'SDK-created ADC client is closed when app.close() is called',
+        () {
+          return runZoned(() async {
+            final app = FirebaseApp.initializeApp(
+              name: 'adc-close-${DateTime.now().microsecondsSinceEpoch}',
+              options: const AppOptions(projectId: projectId),
+            );
+
+            await app.client;
+            await app.close();
+
+            expect(app.isDeleted, isTrue);
+          }, zoneValues: {envSymbol: prodEnv()});
+        },
+        timeout: const Timeout(Duration(seconds: 30)),
+      );
     }, tags: 'prod');
 
     group('_createDefaultClient – service account path', () {
@@ -90,25 +94,29 @@ void main() {
       final refreshTokenFile =
           Platform.environment['FIREBASE_REFRESH_TOKEN_CREDENTIALS'];
 
-      test('creates an authenticated client via refresh token credential', () {
-        return runZoned(() async {
-          final credential = Credential.fromRefreshToken(
-            File(refreshTokenFile!),
-          );
-
-          final app = FirebaseApp.initializeApp(
-            name: 'rt-client-${DateTime.now().microsecondsSinceEpoch}',
-            options: AppOptions(projectId: projectId, credential: credential),
-          );
-
-          try {
-            final client = await app.client;
-            expect(client, isNotNull);
-          } finally {
-            await app.close();
-          }
-        }, zoneValues: {envSymbol: prodEnv()});
-      }, timeout: const Timeout(Duration(seconds: 30)));
+      test(
+        'creates an authenticated client via refresh token credential',
+        () {
+          return runZoned(() async {
+            final credential = Credential.fromRefreshToken(
+              File(refreshTokenFile!),
+            );
+
+            final app = FirebaseApp.initializeApp(
+              name: 'rt-client-${DateTime.now().microsecondsSinceEpoch}',
+              options: AppOptions(projectId: projectId, credential: credential),
+            );
+
+            try {
+              final client = await app.client;
+              expect(client, isNotNull);
+            } finally {
+              await app.close();
+            }
+          }, zoneValues: {envSymbol: prodEnv()});
+        },
+        timeout: const Timeout(Duration(seconds: 30)),
+      );
 
       test(
         'SDK-created refresh token client is closed when app.close() is called',

packages/firebase_admin_sdk/test/integration/app/firebase_app_test.dart

@@ -89,15 +89,19 @@ void main() {
           timeout: const Timeout(Duration(seconds: 30)),
         );
 
-        test('closes multiple services concurrently without error', () async {
-          app.firestore();
-          app.messaging();
-          app.securityRules();
+        test(
+          'closes multiple services concurrently without error',
+          () async {
+            app.firestore();
+            app.messaging();
+            app.securityRules();
 
-          await expectLater(app.close(), completes);
+            await expectLater(app.close(), completes);
 
-          expect(app.isDeleted, isTrue);
-        }, timeout: const Timeout(Duration(seconds: 30)));
+            expect(app.isDeleted, isTrue);
+          },
+          timeout: const Timeout(Duration(seconds: 30)),
+        );
       },
       skip: Environment.isFirestoreEmulatorEnabled()
           ? false
@@ -118,26 +122,30 @@ void main() {
           emulatorEnv.remove(Environment.googleApplicationCredentials);
         });
 
-        test('initialises Auth, creates a user, then closes cleanly', () async {
-          await runZoned(zoneValues: {envSymbol: emulatorEnv}, () async {
-            final app = FirebaseApp.initializeApp(
-              name: 'auth-lifecycle-${DateTime.now().millisecondsSinceEpoch}',
-              options: const AppOptions(projectId: projectId),
-            );
+        test(
+          'initialises Auth, creates a user, then closes cleanly',
+          () async {
+            await runZoned(zoneValues: {envSymbol: emulatorEnv}, () async {
+              final app = FirebaseApp.initializeApp(
+                name: 'auth-lifecycle-${DateTime.now().millisecondsSinceEpoch}',
+                options: const AppOptions(projectId: projectId),
+              );
 
-            final auth = Auth.internal(app);
+              final auth = Auth.internal(app);
 
-            final user = await auth.createUser(
-              CreateRequest(email: 'lifecycle-test@example.com'),
-            );
-            expect(user.email, 'lifecycle-test@example.com');
+              final user = await auth.createUser(
+                CreateRequest(email: 'lifecycle-test@example.com'),
+              );
+              expect(user.email, 'lifecycle-test@example.com');
 
-            await auth.deleteUser(user.uid);
+              await auth.deleteUser(user.uid);
 
-            await app.close();
-            expect(app.isDeleted, isTrue);
-          });
-        }, timeout: const Timeout(Duration(seconds: 30)));
+              await app.close();
+              expect(app.isDeleted, isTrue);
+            });
+          },
+          timeout: const Timeout(Duration(seconds: 30)),
+        );
       },
       skip: Environment.isAuthEmulatorEnabled()
           ? false

packages/firebase_admin_sdk/test/integration/security_rules/security_rules_prod_test.dart

@@ -123,27 +123,31 @@ void main() {
       expect(after.name, ruleset.name);
     });
 
-    test('storage release flow', () async {
-      const bucket = 'dart-firebase-admin.appspot.com';
-
-      // Create and release a new ruleset from source
-      final newRuleset = await securityRules.releaseStorageRulesetFromSource(
-        simpleStorageContent,
-        bucket,
-      );
-      createdRulesets.add(newRuleset.name);
-
-      expect(newRuleset.name, isNotEmpty);
-      expect(newRuleset.source.length, 1);
-      expect(newRuleset.source.single.name, 'storage.rules');
-      expect(newRuleset.source.single.content, simpleStorageContent);
-
-      // Verify it was applied by getting the current ruleset
-      final after = await securityRules.getStorageRuleset(bucket);
-      expect(after.name, newRuleset.name);
-      expect(after.source.length, 1);
-      expect(after.source.single.content, simpleStorageContent);
-    }, skip: 'Requires Storage bucket to be configured in Firebase project');
+    test(
+      'storage release flow',
+      () async {
+        const bucket = 'dart-firebase-admin.appspot.com';
+
+        // Create and release a new ruleset from source
+        final newRuleset = await securityRules.releaseStorageRulesetFromSource(
+          simpleStorageContent,
+          bucket,
+        );
+        createdRulesets.add(newRuleset.name);
+
+        expect(newRuleset.name, isNotEmpty);
+        expect(newRuleset.source.length, 1);
+        expect(newRuleset.source.single.name, 'storage.rules');
+        expect(newRuleset.source.single.content, simpleStorageContent);
+
+        // Verify it was applied by getting the current ruleset
+        final after = await securityRules.getStorageRuleset(bucket);
+        expect(after.name, newRuleset.name);
+        expect(after.source.length, 1);
+        expect(after.source.single.content, simpleStorageContent);
+      },
+      skip: 'Requires Storage bucket to be configured in Firebase project',
+    );
 
     group('Error Handling', () {
       test(