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;
@@ -4273,6 +4274,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
@@ -4289,6 +4291,7 @@ public function updateDocuments(
  array $queries = [],
  int $batchSize = self::INSERT_BATCH_SIZE,
  ?callable $onNext = null,
+ ?callable $onError = null,
  ): int {
  if ($updates->isEmpty()) {
  return 0;
@@ -4389,30 +4392,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,
@@ -4423,7 +4425,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

@@ -3427,6 +3427,7 @@ public function testUpdateDocument(Document $document): Document
  return $document;
  }
 
+
  /**
  * @depends testUpdateDocument
  */
@@ -3691,6 +3692,102 @@ public function testUpdateDocuments(): void
  Authorization::cleanRoles();
  Authorization::setRole(Role::any()->toString());
  }
+
+ public function testUpdateDocumentsWithCallbackSupport(): void
+ {
+ /** @var Database $database */
+ $database = static::getDatabase();
+
+ if (!$database->getAdapter()->getSupportForBatchOperations()) {
+ $this->expectNotToPerformAssertions();
+ return;
+ }
+
+ $collection = 'update_callback';
+ Authorization::cleanRoles();
+ Authorization::setRole(Role::any()->toString());
+
+ $database->createCollection($collection, attributes: [
+ new Document([
+ '$id' => ID::custom('string'),
+ 'type' => Database::VAR_STRING,
+ 'format' => '',
+ 'size' => 100,
+ 'signed' => true,
+ 'required' => false,
+ 'default' => null,
+ 'array' => false,
+ 'filters' => [],
+ ]),
+ new Document([
+ '$id' => ID::custom('integer'),
+ 'type' => Database::VAR_INTEGER,
+ 'format' => '',
+ 'size' => 10000,
+ 'signed' => true,
+ 'required' => false,
+ 'default' => null,
+ 'array' => false,
+ 'filters' => [],
+ ]),
+ ], permissions: [
+ Permission::read(Role::any()),
+ Permission::create(Role::any()),
+ Permission::update(Role::any()),
+ Permission::delete(Role::any())
+ ], documentSecurity: false);
+
+ for ($i = 0; $i < 10; $i++) {
+ $database->createDocument($collection, new Document([
+ '$id' => 'doc' . $i,
+ 'string' => 'text📝 ' . $i,
+ 'integer' => $i
+ ]));
+ }
+ // Test onNext is throwing the error without the onError
+ // a non existent document to test the error thrown
+ try {
+ $results = [];
+ $count = $database->updateDocuments($collection, new Document([
+ 'string' => 'text📝 updated',
+ ]), [
+ Query::greaterThanEqual('integer', 100),
+ ], onNext: function ($doc) use (&$results) {
+ $results[] = $doc;
+ throw new Exception("Error thrown to test that update doesn't stop 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());
+ }
+
+ // Test Update half of the documents
+ $results = [];
+ $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 that update doesn't stop and error is caught");
+ }, onError:function ($e) {
+ $this->assertInstanceOf(Exception::class, $e);
+ $this->assertEquals("Error thrown to test that update doesn't stop and error is caught", $e->getMessage());
+ });
+
+ $this->assertEquals(5, $count);
+
+ foreach ($results as $document) {
+ $this->assertEquals('text📝 updated', $document->getAttribute('string'));
+ }
+
+ $updatedDocuments = $database->find($collection, [
+ Query::greaterThanEqual('integer', 5),
+ ]);
+
+ $this->assertCount(5, $updatedDocuments);
+ }
+
  /**
  * @depends testCreateDocument
  */

tests/e2e/Adapter/Scopes/Relationships/ManyToManyTests.php

@@ -6,6 +6,7 @@
 use Utopia\Database\Database;
 use Utopia\Database\Document;
 use Utopia\Database\Exception\Restricted as RestrictedException;
+use Utopia\Database\Exception\Structure;
 use Utopia\Database\Helpers\ID;
 use Utopia\Database\Helpers\Permission;
 use Utopia\Database\Helpers\Role;
@@ -1595,4 +1596,89 @@ public function testDeleteBulkDocumentsManyToManyRelationship(): void
  $this->getDatabase()->deleteDocuments('bulk_delete_person_m2m');
  $this->assertCount(0, $this->getDatabase()->find('bulk_delete_person_m2m'));
  }
+ public function testUpdateParentAndChild_ManyToMany(): void
+ {
+ /** @var Database $database */
+ $database = static::getDatabase();
+
+ if (
+ !$database->getAdapter()->getSupportForRelationships() ||
+ !$database->getAdapter()->getSupportForBatchOperations()
+ ) {
+ $this->expectNotToPerformAssertions();
+ return;
+ }
+
+ $parentCollection = 'parent_combined_m2m';
+ $childCollection = 'child_combined_m2m';
+
+ $database->createCollection($parentCollection);
+ $database->createCollection($childCollection);
+
+ $database->createAttribute($parentCollection, 'name', Database::VAR_STRING, 255, true);
+ $database->createAttribute($childCollection, 'name', Database::VAR_STRING, 255, true);
+ $database->createAttribute($childCollection, 'parentNumber', Database::VAR_INTEGER, 0, false);
+
+
+ $database->createRelationship(
+ collection: $parentCollection,
+ relatedCollection: $childCollection,
+ type: Database::RELATION_MANY_TO_MANY,
+ id: 'parentNumber'
+ );
+
+ $database->createDocument($parentCollection, new Document([
+ '$id' => 'parent1',
+ '$permissions' => [
+ Permission::read(Role::any()),
+ Permission::update(Role::any()),
+ Permission::delete(Role::any()),
+ ],
+ 'name' => 'Parent 1',
+ ]));
+
+ $database->createDocument($childCollection, new Document([
+ '$id' => 'child1',
+ '$permissions' => [
+ Permission::read(Role::any()),
+ Permission::update(Role::any()),
+ Permission::delete(Role::any()),
+ ],
+ 'name' => 'Child 1',
+ 'parentNumber' => null,
+ ]));
+
+ $database->updateDocuments(
+ $parentCollection,
+ new Document(['name' => 'Parent 1 Updated']),
+ [Query::equal('$id', ['parent1'])]
+ );
+
+ $parentDoc = $database->getDocument($parentCollection, 'parent1');
+ $this->assertEquals('Parent 1 Updated', $parentDoc->getAttribute('name'), 'Parent should be updated');
+
+ $childDoc = $database->getDocument($childCollection, 'child1');
+ $this->assertEquals('Child 1', $childDoc->getAttribute('name'), 'Child should remain unchanged');
+
+ // invalid update to child
+ try {
+ $database->updateDocuments(
+ $childCollection,
+ new Document(['parentNumber' => 'not-a-number']),
+ [Query::equal('$id', ['child1'])]
+ );
+ $this->fail('Expected exception was not thrown for invalid parentNumber type');
+ } catch (\Throwable $e) {
+ $this->assertInstanceOf(Structure::class, $e);
+ }
+
+ // parent remains unaffected
+ $parentDocAfter = $database->getDocument($parentCollection, 'parent1');
+ $this->assertEquals('Parent 1 Updated', $parentDocAfter->getAttribute('name'), 'Parent should not be affected by failed child update');
+
+ $database->deleteCollection($parentCollection);
+ $database->deleteCollection($childCollection);
+ }
+
+
 }