PHPLIB-1419 Encode Agg builder objects in Collection methods

Ignore static analysis issues

PHPLIB-1419 Fix BC break

Author: GromNaN

psalm-baseline.xml

@@ -223,6 +223,11 @@
     <MixedPropertyTypeCoercion>
       <code><![CDATA[$options['builderEncoder'] ?? new BuilderEncoder()]]></code>
     </MixedPropertyTypeCoercion>
+    <PossiblyInvalidArgument>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+      <code><![CDATA[$pipeline]]></code>
+    </PossiblyInvalidArgument>
   </file>
   <file src="src/Command/ListCollections.php">
     <MixedAssignment>

src/Collection.php

@@ -23,6 +23,7 @@
 use MongoDB\BSON\JavascriptInterface;
 use MongoDB\BSON\PackedArray;
 use MongoDB\Builder\BuilderEncoder;
+use MongoDB\Builder\Pipeline;
 use MongoDB\Codec\DocumentCodec;
 use MongoDB\Codec\Encoder;
 use MongoDB\Driver\CursorInterface;
@@ -223,6 +224,12 @@ public function __toString()
      */
     public function aggregate(array $pipeline, array $options = [])
     {
+        if (is_builder_pipeline($pipeline)) {
+            $pipeline = new Pipeline(...$pipeline);
+        }
+
+        $pipeline = $this->builderEncoder->encodeIfSupported($pipeline);
+
         $hasWriteStage = is_last_pipeline_operator_write($pipeline);
 
         $options = $this->inheritReadPreference($options);
@@ -1098,6 +1105,11 @@ public function updateSearchIndex(string $name, array|object $definition, array
      */
     public function watch(array $pipeline = [], array $options = [])
     {
+        if (is_builder_pipeline($pipeline)) {
+            $pipeline = new Pipeline(...$pipeline);
+        }
+
+        $pipeline = $this->builderEncoder->encodeIfSupported($pipeline);
         $options = $this->inheritReadOptions($options);
         $options = $this->inheritCodecOrTypeMap($options);
 

src/functions.php

@@ -21,6 +21,7 @@
 use MongoDB\BSON\Document;
 use MongoDB\BSON\PackedArray;
 use MongoDB\BSON\Serializable;
+use MongoDB\Builder\Type\StageInterface;
 use MongoDB\Driver\Exception\RuntimeException as DriverRuntimeException;
 use MongoDB\Driver\Manager;
 use MongoDB\Driver\ReadPreference;
@@ -327,6 +328,38 @@ function is_pipeline(array|object $pipeline, bool $allowEmpty = false): bool
     return true;
 }
 
+/**
+ * Returns whether the argument is a list that contains at least one
+ * {@see StageInterface} object.
+ *
+ * @internal
+ */
+function is_builder_pipeline(array $pipeline): bool
+{
+    if (! $pipeline) {
+        return false;
+    }
+
+    if (! array_is_list($pipeline)) {
+        return false;
+    }
+
+    $result = false;
+    foreach ($pipeline as $stage) {
+        if (! is_array($stage) && ! is_object($stage)) {
+            return false;
+        }
+
+        if ($stage instanceof StageInterface) {
+            $result = true;
+        } elseif (! is_first_key_operator($stage)) {
+            return false;
+        }
+    }
+
+    return $result;
+}
+
 /**
  * Returns whether we are currently in a transaction.
  *

tests/Collection/BuilderCollectionFunctionalTest.php

@@ -2,10 +2,13 @@
 
 namespace MongoDB\Tests\Collection;
 
+use MongoDB\Builder\Expression;
 use MongoDB\Builder\Pipeline;
 use MongoDB\Builder\Query;
 use MongoDB\Builder\Stage;
 
+use function iterator_to_array;
+
 class BuilderCollectionFunctionalTest extends FunctionalTestCase
 {
     public function setUp(): void
@@ -17,7 +20,18 @@ public function setUp(): void
 
     public function testAggregate(): void
     {
-        $this->markTestSkipped('Not supported yet');
+        $this->collection->insertMany([['x' => 10], ['x' => 10], ['x' => 10]]);
+        $pipeline = new Pipeline(
+            Stage::bucketAuto(
+                groupBy: Expression::intFieldPath('x'),
+                buckets: 2,
+            ),
+        );
+        // Extract the list of stages for arg type restriction
+        $pipeline = iterator_to_array($pipeline);
+
+        $results = $this->collection->aggregate($pipeline)->toArray();
+        $this->assertCount(2, $results);
     }
 
     public function testBulkWriteDeleteMany(): void
@@ -245,6 +259,24 @@ public function testUpdateManyWithPipeline(): void
 
     public function testWatch(): void
     {
-        $this->markTestSkipped('Not supported yet');
+        $this->skipIfChangeStreamIsNotSupported();
+
+        if ($this->isShardedCluster()) {
+            $this->markTestSkipped('Test does not apply on sharded clusters: need more than a single getMore call on the change stream.');
+        }
+
+        $pipeline = new Pipeline(
+            Stage::match(operationType: Query::eq('insert')),
+        );
+        // Extract the list of stages for arg type restriction
+        $pipeline = iterator_to_array($pipeline);
+
+        $changeStream = $this->collection->watch($pipeline);
+        $changeStream->rewind();
+        $this->assertNull($changeStream->current());
+        $this->collection->insertOne(['x' => 3]);
+        $changeStream->next();
+        $this->assertTrue($changeStream->valid());
+        $this->assertEquals('insert', $changeStream->current()->operationType);
     }
 }

tests/FunctionsTest.php

@@ -4,6 +4,8 @@
 
 use MongoDB\BSON\Document;
 use MongoDB\BSON\PackedArray;
+use MongoDB\Builder\Stage\LimitStage;
+use MongoDB\Builder\Stage\MatchStage;
 use MongoDB\Driver\WriteConcern;
 use MongoDB\Model\BSONArray;
 use MongoDB\Model\BSONDocument;
@@ -12,6 +14,7 @@
 use function MongoDB\apply_type_map_to_document;
 use function MongoDB\create_field_path_type_map;
 use function MongoDB\document_to_array;
+use function MongoDB\is_builder_pipeline;
 use function MongoDB\is_first_key_operator;
 use function MongoDB\is_last_pipeline_operator_write;
 use function MongoDB\is_mapreduce_output_inline;
@@ -311,6 +314,22 @@ public function providePipelines(): array
         ];
     }
 
+    /** @dataProvider provideStagePipelines */
+    public function testIsBuilderPipeline($expected, $pipeline): void
+    {
+        $this->assertSame($expected, is_builder_pipeline($pipeline));
+    }
+
+    public function provideStagePipelines(): iterable
+    {
+        yield 'empty array' => [false, []];
+        yield 'array of arrays' => [false, [['$match' => ['x' => 1]]]];
+        yield 'map of stages' => [false, [1 => new MatchStage([])]];
+        yield 'stages' => [true, [new MatchStage([]), new LimitStage(1)]];
+        yield 'stages and operators' => [true, [new MatchStage([]), ['$limit' => 1]]];
+        yield 'stages and invalid' => [false, [new MatchStage([]), ['foo' => 'bar']]];
+    }
+
     /** @dataProvider provideWriteConcerns */
     public function testIsWriteConcernAcknowledged($expected, WriteConcern $writeConcern): void
     {