Add createOrUpdateDocuments method and filter hooks for document upserts

Author: ArnabChatterjee20k

src/Database/Mirror.php

@@ -781,6 +781,75 @@ function ($doc) use ($onNext, &$modified) {
         return $modified;
     }
 
+    public function createOrUpdateDocuments(string $collection, array $documents, int $batchSize = Database::INSERT_BATCH_SIZE, callable|null $onNext = null): int
+    {
+        $modified = 0;
+        $this->source->createOrUpdateDocuments(
+            $collection,
+            $documents,
+            $batchSize,
+            function ($doc) use ($onNext, &$modified) {
+                $onNext && $onNext($doc);
+                $modified++;
+            }
+        );
+
+        if (
+            \in_array($collection, self::SOURCE_ONLY_COLLECTIONS)
+            || $this->destination === null
+        ) {
+            return $modified;
+        }
+
+        $upgrade = $this->silent(fn () => $this->getUpgradeStatus($collection));
+        if ($upgrade === null || $upgrade->getAttribute('status', '') !== 'upgraded') {
+            return $modified;
+        }
+
+        try {
+            $clones = [];
+
+            foreach ($documents as $document) {
+                $clone = clone $document;
+
+                foreach ($this->writeFilters as $filter) {
+                    $clone = $filter->beforeCreateOrUpdateDocument(
+                        source: $this->source,
+                        destination: $this->destination,
+                        collectionId: $collection,
+                        document: $clone,
+                    );
+                }
+
+                $clones[] = $clone;
+            }
+
+            $this->destination->withPreserveDates(
+                fn () =>
+                $this->destination->createOrUpdateDocuments(
+                    $collection,
+                    $clones,
+                    $batchSize,
+                    null,
+                )
+            );
+
+            foreach ($clones as $clone) {
+                foreach ($this->writeFilters as $filter) {
+                    $filter->afterCreateOrUpdateDocument(
+                        source: $this->source,
+                        destination: $this->destination,
+                        collectionId: $collection,
+                        document: $clone,
+                    );
+                }
+            }
+        } catch (\Throwable $err) {
+            $this->logError('createDocuments', $err);
+        }
+        return $modified;
+    }
+
     public function deleteDocument(string $collection, string $id): bool
     {
         $result = $this->source->deleteDocument($collection, $id);

src/Database/Mirroring/Filter.php

@@ -359,4 +359,40 @@ public function afterDeleteDocuments(
         array $queries
     ): void {
     }
+
+    /**
+     * Called before document is upserted in the destination database
+     *
+     * @param Database $source
+     * @param Database $destination
+     * @param string $collectionId
+     * @param Document $document
+     * @return Document
+     */
+    public function beforeCreateOrUpdateDocument(
+        Database $source,
+        Database $destination,
+        string $collectionId,
+        Document $document,
+    ): Document {
+        return $document;
+    }
+
+    /**
+     * Called after document is upserted in the destination database
+     *
+     * @param Database $source
+     * @param Database $destination
+     * @param string $collectionId
+     * @param Document $document
+     * @return Document
+     */
+    public function afterCreateOrUpdateDocument(
+        Database $source,
+        Database $destination,
+        string $collectionId,
+        Document $document,
+    ): Document {
+        return $document;
+    }
 }

tests/e2e/Adapter/Scopes/DocumentTests.php

@@ -5788,4 +5788,60 @@ public function testUpsertDateOperations(): void
         $database->setPreserveDates(false);
         $database->deleteCollection($collection);
     }
+
+    public function testUpdateDocumentsCount(): void
+    {
+        /** @var Database $database */
+        $database = static::getDatabase();
+
+        if (!$database->getAdapter()->getSupportForUpserts()) {
+            return;
+        }
+
+        $collectionName = "update_count";
+        $database->createCollection($collectionName);
+
+        $database->createAttribute($collectionName, 'key', Database::VAR_STRING, 60, false);
+        $database->createAttribute($collectionName, 'value', Database::VAR_STRING, 60, false);
+
+        $permissions = [Permission::read(Role::any()), Permission::write(Role::any()),Permission::update(Role::any())];
+
+        $docs =  [
+            new Document([
+                '$id' => 'bulk_upsert1',
+                '$permissions' => $permissions,
+                'key' => 'bulk_upsert1_initial',
+            ]),
+            new Document([
+                '$id' => 'bulk_upsert2',
+                '$permissions' => $permissions,
+                'key' => 'bulk_upsert2_initial',
+            ]),
+            new Document([
+                '$id' => 'bulk_upsert3',
+                '$permissions' => $permissions,
+                'key' => 'bulk_upsert3_initial',
+            ]),
+            new Document([
+                '$id' => 'bulk_upsert4',
+                '$permissions' => $permissions,
+                'key' => 'bulk_upsert4_initial'
+            ])
+        ];
+        $upsertUpdateResults = [];
+        $count = $database->createOrUpdateDocuments($collectionName, $docs, onNext: function ($doc) use (&$upsertUpdateResults) {
+            $upsertUpdateResults[] = $doc;
+        });
+        $this->assertCount(4, $upsertUpdateResults);
+        $this->assertEquals(4, $count);
+
+        $updates = new Document(['value' => 'test']);
+        $newDocs = [];
+        $count = $database->updateDocuments($collectionName, $updates, onNext:function ($doc) use (&$newDocs) {
+            $newDocs[] = $doc;
+        });
+
+        $this->assertCount(4, $newDocs);
+        $this->assertEquals(4, $count);
+    }
 }