Add error handling for document updates with onError callback

src/Database/Database.php

@@ -3,6 +3,7 @@
 namespace Utopia\Database;
 
 use Exception;
+use Throwable;
 use Utopia\Cache\Cache;
 use Utopia\CLI\Console;
 use Utopia\Database\Exception as DatabaseException;
@@ -4302,6 +4303,7 @@ public function updateDocument(string $collection, string $id, Document $documen
      * @param array<Query> $queries
      * @param int $batchSize
      * @param callable|null $onNext
+     * @param callable|null $onError
      * @return int
      * @throws AuthorizationException
      * @throws ConflictException
@@ -4318,6 +4320,7 @@ public function updateDocuments(
         array $queries = [],
         int $batchSize = self::INSERT_BATCH_SIZE,
         ?callable $onNext = null,
+        ?callable $onError = null,
     ): int {
         if ($updates->isEmpty()) {
             return 0;
@@ -4418,30 +4421,29 @@ public function updateDocuments(
                 break;
             }
 
-            foreach ($batch as &$document) {
-                $new = new Document(\array_merge($document->getArrayCopy(), $updates->getArrayCopy()));
+            $this->withTransaction(function () use ($collection, $updates, &$batch) {
+                foreach ($batch as &$document) {
+                    $new = new Document(\array_merge($document->getArrayCopy(), $updates->getArrayCopy()));
 
-                if ($this->resolveRelationships) {
-                    $this->silent(fn () => $this->updateDocumentRelationships($collection, $document, $new));
-                }
-
-                $document = $new;
+                    if ($this->resolveRelationships) {
+                        $this->silent(fn () => $this->updateDocumentRelationships($collection, $document, $new));
+                    }
 
-                // Check if document was updated after the request timestamp
-                try {
-                    $oldUpdatedAt = new \DateTime($document->getUpdatedAt());
-                } catch (Exception $e) {
-                    throw new DatabaseException($e->getMessage(), $e->getCode(), $e);
-                }
+                    $document = $new;
 
-                if (!is_null($this->timestamp) && $oldUpdatedAt > $this->timestamp) {
-                    throw new ConflictException('Document was updated after the request timestamp');
-                }
+                    // Check if document was updated after the request timestamp
+                    try {
+                        $oldUpdatedAt = new \DateTime($document->getUpdatedAt());
+                    } catch (Exception $e) {
+                        throw new DatabaseException($e->getMessage(), $e->getCode(), $e);
+                    }
 
-                $document = $this->encode($collection, $document);
-            }
+                    if (!is_null($this->timestamp) && $oldUpdatedAt > $this->timestamp) {
+                        throw new ConflictException('Document was updated after the request timestamp');
+                    }
 
-            $this->withTransaction(function () use ($collection, $updates, $batch) {
+                    $document = $this->encode($collection, $document);
+                }
                 $this->adapter->updateDocuments(
                     $collection->getId(),
                     $updates,
@@ -4452,7 +4454,11 @@ public function updateDocuments(
             foreach ($batch as $doc) {
                 $this->purgeCachedDocument($collection->getId(), $doc->getId());
                 $doc = $this->decode($collection, $doc);
-                $onNext && $onNext($doc);
+                try {
+                    $onNext && $onNext($doc);
+                } catch (Throwable $th) {
+                    $onError ? $onError($th) : throw $th;
+                }
                 $modified++;
             }
 

src/Database/Mirror.php

@@ -713,6 +713,7 @@ public function updateDocuments(
         array $queries = [],
         int $batchSize = self::INSERT_BATCH_SIZE,
         ?callable $onNext = null,
+        ?callable $onError = null,
     ): int {
         $modified = 0;
 
@@ -724,7 +725,8 @@ public function updateDocuments(
             function ($doc) use ($onNext, &$modified) {
                 $onNext && $onNext($doc);
                 $modified++;
-            }
+            },
+            $onError
         );
 
         if (

tests/e2e/Adapter/Scopes/DocumentTests.php

@@ -3550,12 +3550,29 @@ public function testUpdateDocuments(): void
 
         // Test Update half of the documents
         $results = [];
+        try {
+            $count = $database->updateDocuments($collection, new Document([
+                'string' => 'text📝 updated',
+            ]), [
+                Query::lessThan('integer', -1),
+            ], onNext: function ($doc) use (&$results) {
+                $results[] = $doc;
+                throw new Exception("Error thrown to test update doesn't stopped and error is caught");
+            });
+        } catch (Exception $e) {
+            $this->assertInstanceOf(Exception::class, $e);
+            $this->assertEquals("Error thrown to test that update doesn't stop and error is caught", $e->getMessage());
+        }
         $count = $database->updateDocuments($collection, new Document([
             'string' => 'text📝 updated',
         ]), [
             Query::greaterThanEqual('integer', 5),
         ], onNext: function ($doc) use (&$results) {
             $results[] = $doc;
+            throw new Exception("Error thrown to test update doesn't stopped and error is caught");
+        }, onError:function ($e) {
+            $this->assertInstanceOf(Exception::class, $e);
+            $this->assertEquals("Error thrown to test update doesn't stopped and error is caught", $e->getMessage());
         });
 
         $this->assertEquals(5, $count);
@@ -3588,7 +3605,16 @@ public function testUpdateDocuments(): void
         // Test Update all documents
         $this->assertEquals(10, $database->updateDocuments($collection, new Document([
             'string' => 'text📝 updated all',
-        ])));
+        ]), onNext: function () {
+            throw new DatabaseException("Error thrown to test update doesn't stopped and error is caught");
+        }, onError:function ($e) {
+            if ($e instanceof DatabaseException) {
+                $this->assertInstanceOf(DatabaseException::class, $e);
+                $this->assertEquals("Error thrown to test update doesn't stopped and error is caught", $e->getMessage());
+            } else {
+                $this->fail("Caught value is not an Exception.");
+            }
+        }));
 
         $updatedDocuments = $database->find($collection);
 

tests/e2e/Adapter/Scopes/RelationshipTests.php

@@ -1791,7 +1791,12 @@ public function testUpdateDocumentsRelationships(): void
 
         $this->getDatabase()->updateDocuments('testUpdateDocumentsRelationships1', new Document([
             'string' => 'text📝 updated',
-        ]));
+        ]), onNext: function () {
+            throw new Exception("Error thrown to test update doesn't stopped and error is caught");
+        }, onError:function ($e) {
+            $this->assertInstanceOf(Exception::class, $e);
+            $this->assertEquals("Error thrown to test update doesn't stopped and error is caught", $e->getMessage());
+        });
 
         $document = $this->getDatabase()->findOne('testUpdateDocumentsRelationships1');
 
@@ -1829,11 +1834,21 @@ public function testUpdateDocumentsRelationships(): void
 
         $this->getDatabase()->updateDocuments('testUpdateDocumentsRelationships2', new Document([
             'testUpdateDocumentsRelationships1' => null
-        ]));
+        ]), onNext: function () {
+            throw new Exception("Error thrown to test update doesn't stopped and error is caught");
+        }, onError:function ($e) {
+            $this->assertInstanceOf(Exception::class, $e);
+            $this->assertEquals("Error thrown to test update doesn't stopped and error is caught", $e->getMessage());
+        });
 
         $this->getDatabase()->updateDocuments('testUpdateDocumentsRelationships2', new Document([
             'testUpdateDocumentsRelationships1' => 'doc1'
-        ]));
+        ]), onNext: function () {
+            throw new Exception("Error thrown to test update doesn't stopped and error is caught");
+        }, onError:function ($e) {
+            $this->assertInstanceOf(Exception::class, $e);
+            $this->assertEquals("Error thrown to test update doesn't stopped and error is caught", $e->getMessage());
+        });
 
         $documents = $this->getDatabase()->find('testUpdateDocumentsRelationships2');