diff --git a/src/Database/Database.php b/src/Database/Database.php
index 1205eb0a2..acf79db0e 100644
--- a/src/Database/Database.php
+++ b/src/Database/Database.php
@@ -3455,32 +3455,29 @@ public function createDocument(string $collection, Document $document): Document
      * @param string $collection
      * @param array<Document> $documents
      * @param int $batchSize
-     * @return array<Document>
+     * @param callable|null $onNext
+     * @return int
      * @throws AuthorizationException
      * @throws StructureException
-     * @throws NotFoundException
      * @throws \Throwable
+     * @throws Exception
      */
     public function createDocuments(
         string $collection,
         array $documents,
         int $batchSize = self::INSERT_BATCH_SIZE,
-    ): array {
+        ?callable $onNext = null,
+    ): int {
         if (!$this->adapter->getSharedTables() && $this->adapter->getTenantPerDocument()) {
             throw new DatabaseException('Shared tables must be enabled if tenant per document is enabled.');
         }
 
         if (empty($documents)) {
-            return [];
+            return 0;
         }
 
-        $collection = $this->silent(fn () => $this->getCollection($collection));
-
         $batchSize = \min(Database::INSERT_BATCH_SIZE, \max(1, $batchSize));
-
-        /**
-         * Check collection exist
-         */
+        $collection = $this->silent(fn () => $this->getCollection($collection));
         if ($collection->getId() !== self::METADATA) {
             $authorization = new Authorization(self::PERMISSION_CREATE);
             if (!$authorization->isValid($collection->getCreate())) {
@@ -3489,8 +3486,9 @@ public function createDocuments(
         }
 
         $time = DateTime::now();
+        $modified = 0;
 
-        foreach ($documents as $key => $document) {
+        foreach ($documents as &$document) {
             $createdAt = $document->getCreatedAt();
             $updatedAt = $document->getUpdatedAt();
 
@@ -3524,34 +3522,28 @@ public function createDocuments(
             if ($this->resolveRelationships) {
                 $document = $this->silent(fn () => $this->createDocumentRelationships($collection, $document));
             }
-
-            $documents[$key] = $document;
         }
 
-        $documents = $this->withTransaction(function () use ($collection, $documents, $batchSize) {
-            $stack = [];
-
-            foreach (\array_chunk($documents, $batchSize) as $chunk) {
-                \array_push($stack, ...$this->adapter->createDocuments($collection->getId(), $chunk));
-            }
-
-            return $stack;
-        });
+        foreach (\array_chunk($documents, $batchSize) as $chunk) {
+            $batch = $this->withTransaction(function () use ($collection, $chunk) {
+                return $this->adapter->createDocuments($collection->getId(), $chunk);
+            });
 
-        foreach ($documents as $key => $document) {
-            if ($this->resolveRelationships) {
-                $document = $this->silent(fn () => $this->populateDocumentRelationships($collection, $document));
+            foreach ($batch as $doc) {
+                if ($this->resolveRelationships) {
+                    $doc = $this->silent(fn () => $this->populateDocumentRelationships($collection, $doc));
+                }
+                $onNext && $onNext($doc);
+                $modified++;
             }
-
-            $documents[$key] = $this->decode($collection, $document);
         }
 
         $this->trigger(self::EVENT_DOCUMENTS_CREATE, new Document([
             '$collection' => $collection->getId(),
-            'modified' => count($documents)
+            'modified' => $modified
         ]));
 
-        return $documents;
+        return $modified;
     }
 
     /**
@@ -4085,28 +4077,35 @@ public function updateDocument(string $collection, string $id, Document $documen
      * @param Document $updates
      * @param array<Query> $queries
      * @param int $batchSize
-     *
-     * @return array<Document>
-     *
+     * @param callable|null $onNext
+     * @return int
      * @throws AuthorizationException
-     * @throws DatabaseException
+     * @throws ConflictException
+     * @throws DuplicateException
+     * @throws QueryException
+     * @throws StructureException
+     * @throws TimeoutException
+     * @throws \Throwable
+     * @throws Exception
      */
-    public function updateDocuments(string $collection, Document $updates, array $queries = [], int $batchSize = self::INSERT_BATCH_SIZE): array
-    {
+    public function updateDocuments(
+        string $collection,
+        Document $updates,
+        array $queries = [],
+        int $batchSize = self::INSERT_BATCH_SIZE,
+        ?callable $onNext = null,
+    ): int {
         if ($updates->isEmpty()) {
-            return [];
+            return 0;
         }
 
         $batchSize = \min(Database::INSERT_BATCH_SIZE, \max(1, $batchSize));
-
         $collection = $this->silent(fn () => $this->getCollection($collection));
-
         if ($collection->isEmpty()) {
             throw new DatabaseException('Collection not found');
         }
 
         $documentSecurity = $collection->getAttribute('documentSecurity', false);
-
         $authorization = new Authorization(self::PERMISSION_UPDATE);
         $skipAuth = $authorization->isValid($collection->getUpdate());
 
@@ -4160,9 +4159,9 @@ public function updateDocuments(string $collection, Document $updates, array $qu
             throw new StructureException($validator->getDescription());
         }
 
-        $documents = [];
         $originalLimit = $limit;
-        $lastDocument = $cursor;
+        $last = $cursor;
+        $modified = 0;
 
         // Resolve and update relationships
         while (true) {
@@ -4176,25 +4175,25 @@ public function updateDocuments(string $collection, Document $updates, array $qu
                 Query::limit($batchSize)
             ];
 
-            if (!empty($lastDocument)) {
-                $new[] = Query::cursorAfter($lastDocument);
+            if (!empty($last)) {
+                $new[] = Query::cursorAfter($last);
             }
 
-            $affectedDocuments = $this->silent(fn () => $this->find(
+            $batch = $this->silent(fn () => $this->find(
                 $collection->getId(),
                 array_merge($new, $queries),
                 forPermission: Database::PERMISSION_UPDATE
             ));
 
-            if (empty($affectedDocuments)) {
+            if (empty($batch)) {
                 break;
             }
 
-            foreach ($affectedDocuments as $document) {
+            foreach ($batch as &$document) {
                 if ($this->resolveRelationships) {
                     $newDocument = new Document(array_merge($document->getArrayCopy(), $updates->getArrayCopy()));
                     $this->silent(fn () => $this->updateDocumentRelationships($collection, $document, $newDocument));
-                    $documents[] = $newDocument;
+                    $document = $newDocument;
                 }
 
                 // Check if document was updated after the request timestamp
@@ -4209,11 +4208,11 @@ public function updateDocuments(string $collection, Document $updates, array $qu
                 }
             }
 
-            $this->withTransaction(function () use ($collection, $updates, $authorization, $skipAuth, $affectedDocuments) {
+            $this->withTransaction(function () use ($collection, $updates, $authorization, $skipAuth, $batch) {
                 $getResults = fn () => $this->adapter->updateDocuments(
                     $collection->getId(),
                     $updates,
-                    $affectedDocuments
+                    $batch
                 );
 
                 $skipAuth
@@ -4221,31 +4220,34 @@ public function updateDocuments(string $collection, Document $updates, array $qu
                     : $getResults();
             });
 
-            foreach ($documents as $document) {
+            foreach ($batch as $doc) {
                 if ($this->getSharedTables() && $this->getTenantPerDocument()) {
-                    $this->withTenant($document->getTenant(), function () use ($collection, $document) {
-                        $this->purgeCachedDocument($collection->getId(), $document->getId());
+                    $this->withTenant($doc->getTenant(), function () use ($collection, $doc) {
+                        $this->purgeCachedDocument($collection->getId(), $doc->getId());
                     });
                 } else {
-                    $this->purgeCachedDocument($collection->getId(), $document->getId());
+                    $this->purgeCachedDocument($collection->getId(), $doc->getId());
                 }
+
+                $onNext && $onNext($doc);
+                $modified++;
             }
 
-            if (count($affectedDocuments) < $batchSize) {
+            if (count($batch) < $batchSize) {
                 break;
-            } elseif ($originalLimit && count($documents) == $originalLimit) {
+            } elseif ($originalLimit && $modified == $originalLimit) {
                 break;
             }
 
-            $lastDocument = end($affectedDocuments);
+            $last = \end($batch);
         }
 
         $this->trigger(self::EVENT_DOCUMENTS_UPDATE, new Document([
             '$collection' => $collection->getId(),
-            'modified' => count($documents)
+            'modified' => $modified
         ]));
 
-        return $documents;
+        return $modified;
     }
 
     /**
@@ -4646,19 +4648,22 @@ private function getJunctionCollection(Document $collection, Document $relatedCo
      * @param string $collection
      * @param array<Document> $documents
      * @param int $batchSize
-     * @return array<Document>
+     * @param callable|null $onNext
+     * @return int
      * @throws StructureException
      * @throws \Throwable
      */
     public function createOrUpdateDocuments(
         string $collection,
         array $documents,
-        int $batchSize = self::INSERT_BATCH_SIZE
-    ): array {
+        int $batchSize = self::INSERT_BATCH_SIZE,
+        ?callable $onNext = null,
+    ): int {
         return $this->createOrUpdateDocumentsWithIncrease(
             $collection,
             '',
             $documents,
+            $onNext,
             $batchSize
         );
     }
@@ -4669,8 +4674,9 @@ public function createOrUpdateDocuments(
      * @param string $collection
      * @param string $attribute
      * @param array<Document> $documents
+     * @param callable|null $onNext
      * @param int $batchSize
-     * @return array<Document>
+     * @return int
      * @throws StructureException
      * @throws \Throwable
      * @throws Exception
@@ -4679,10 +4685,11 @@ public function createOrUpdateDocumentsWithIncrease(
         string $collection,
         string $attribute,
         array $documents,
+        ?callable $onNext = null,
         int $batchSize = self::INSERT_BATCH_SIZE
-    ): array {
+    ): int {
         if (empty($documents)) {
-            return [];
+            return 0;
         }
 
         $batchSize = \min(Database::INSERT_BATCH_SIZE, \max(1, $batchSize));
@@ -4776,7 +4783,7 @@ public function createOrUpdateDocumentsWithIncrease(
             );
         }
 
-        $stack = [];
+        $modified = 0;
 
         foreach (\array_chunk($documents, $batchSize) as $chunk) {
             /**
@@ -4788,31 +4795,30 @@ public function createOrUpdateDocumentsWithIncrease(
                 $chunk
             )));
 
-            foreach ($batch as $document) {
+            foreach ($batch as $doc) {
                 if ($this->resolveRelationships) {
-                    $document = $this->silent(fn () => $this->populateDocumentRelationships($collection, $document));
+                    $doc = $this->silent(fn () => $this->populateDocumentRelationships($collection, $doc));
                 }
 
                 if ($this->getSharedTables() && $this->getTenantPerDocument()) {
-                    $this->withTenant($document->getTenant(), function () use ($collection, $document) {
-                        $this->purgeCachedDocument($collection->getId(), $document->getId());
+                    $this->withTenant($doc->getTenant(), function () use ($collection, $doc) {
+                        $this->purgeCachedDocument($collection->getId(), $doc->getId());
                     });
                 } else {
-                    $this->purgeCachedDocument($collection->getId(), $document->getId());
+                    $this->purgeCachedDocument($collection->getId(), $doc->getId());
                 }
 
-                $stack[] = $document;
+                $onNext && $onNext($doc);
+                $modified++;
             }
         }
 
-        $documents = $stack;
-
         $this->trigger(self::EVENT_DOCUMENTS_UPSERT, new Document([
             '$collection' => $collection->getId(),
-            'modified' => \count($documents)
+            'modified' => $modified,
         ]));
 
-        return $documents;
+        return $modified;
     }
 
     /**
@@ -5451,24 +5457,25 @@ private function deleteCascade(Document $collection, Document $relatedCollection
      * @param string $collection
      * @param array<Query> $queries
      * @param int $batchSize
-     *
-     * @return array<Document>
-     *
+     * @param callable|null $onNext
+     * @return int
      * @throws AuthorizationException
      * @throws DatabaseException
      * @throws RestrictedException
      * @throws \Throwable
      */
-    public function deleteDocuments(string $collection, array $queries = [], int $batchSize = self::DELETE_BATCH_SIZE): array
-    {
+    public function deleteDocuments(
+        string $collection,
+        array $queries = [],
+        int $batchSize = self::DELETE_BATCH_SIZE,
+        ?callable $onNext = null,
+    ): int {
         if ($this->adapter->getSharedTables() && empty($this->adapter->getTenant())) {
             throw new DatabaseException('Missing tenant. Tenant must be set when table sharing is enabled.');
         }
 
         $batchSize = \min(Database::DELETE_BATCH_SIZE, \max(1, $batchSize));
-
         $collection = $this->silent(fn () => $this->getCollection($collection));
-
         if ($collection->isEmpty()) {
             throw new DatabaseException('Collection not found');
         }
@@ -5506,9 +5513,9 @@ public function deleteDocuments(string $collection, array $queries = [], int $ba
             throw new DatabaseException("Cursor document must be from the same Collection.");
         }
 
-        $documents = [];
         $originalLimit = $limit;
-        $lastDocument = $cursor;
+        $last = $cursor;
+        $modified = 0;
 
         while (true) {
             if ($limit && $limit < $batchSize && $limit > 0) {
@@ -5521,27 +5528,26 @@ public function deleteDocuments(string $collection, array $queries = [], int $ba
                 Query::limit($batchSize)
             ];
 
-            if (!empty($lastDocument)) {
-                $new[] = Query::cursorAfter($lastDocument);
+            if (!empty($last)) {
+                $new[] = Query::cursorAfter($last);
             }
 
             /**
-             * @var array<Document> $affectedDocuments
+             * @var array<Document> $batch
              */
-            $affectedDocuments = $this->silent(fn () => $this->find(
+            $batch = $this->silent(fn () => $this->find(
                 $collection->getId(),
                 array_merge($new, $queries),
                 forPermission: Database::PERMISSION_DELETE
             ));
 
-            if (empty($affectedDocuments)) {
+            if (empty($batch)) {
                 break;
             }
 
             $internalIds = [];
             $permissionIds = [];
-            foreach ($affectedDocuments as $document) {
-                $documents[] = $document;
+            foreach ($batch as $document) {
                 $internalIds[] = $document->getInternalId();
                 if (!empty($document->getPermissions())) {
                     $permissionIds[] = $document->getId();
@@ -5573,10 +5579,12 @@ public function deleteDocuments(string $collection, array $queries = [], int $ba
                     $permissionIds
                 );
 
-                $skipAuth ? $authorization->skip($getResults) : $getResults();
+                $skipAuth
+                    ? $authorization->skip($getResults)
+                    : $getResults();
             });
 
-            foreach ($affectedDocuments as $document) {
+            foreach ($batch as $document) {
                 if ($this->getSharedTables() && $this->getTenantPerDocument()) {
                     $this->withTenant($document->getTenant(), function () use ($collection, $document) {
                         $this->purgeCachedDocument($collection->getId(), $document->getId());
@@ -5584,27 +5592,26 @@ public function deleteDocuments(string $collection, array $queries = [], int $ba
                 } else {
                     $this->purgeCachedDocument($collection->getId(), $document->getId());
                 }
+
+                $onNext && $onNext($document);
+                $modified++;
             }
 
-            if (count($affectedDocuments) < $batchSize) {
+            if (count($batch) < $batchSize) {
                 break;
-            } elseif ($originalLimit && count($documents) >= $originalLimit) {
+            } elseif ($originalLimit && $modified >= $originalLimit) {
                 break;
             }
 
-            $lastDocument = end($affectedDocuments);
-        }
-
-        if (empty($documents)) {
-            return [];
+            $last = \end($batch);
         }
 
         $this->trigger(self::EVENT_DOCUMENTS_DELETE, new Document([
             '$collection' => $collection->getId(),
-            'modified' => count($documents)
+            'modified' => $modified
         ]));
 
-        return $documents;
+        return $modified;
     }
 
     /**
diff --git a/src/Database/Mirror.php b/src/Database/Mirror.php
index 6ac448b22..2a8ad4dbe 100644
--- a/src/Database/Mirror.php
+++ b/src/Database/Mirror.php
@@ -553,20 +553,31 @@ public function createDocument(string $collection, Document $document): Document
     public function createDocuments(
         string $collection,
         array $documents,
-        int $batchSize = self::INSERT_BATCH_SIZE
-    ): array {
-        $documents = $this->source->createDocuments($collection, $documents, $batchSize);
+        int $batchSize = self::INSERT_BATCH_SIZE,
+        ?callable $onNext = null,
+    ): int {
+        $modified = 0;
+
+        $this->source->createDocuments(
+            $collection,
+            $documents,
+            $batchSize,
+            function ($doc) use ($onNext, &$modified) {
+                $onNext && $onNext($doc);
+                $modified++;
+            },
+        );
 
         if (
             \in_array($collection, self::SOURCE_ONLY_COLLECTIONS)
             || $this->destination === null
         ) {
-            return $documents;
+            return $modified;
         }
 
         $upgrade = $this->silent(fn () => $this->getUpgradeStatus($collection));
         if ($upgrade === null || $upgrade->getAttribute('status', '') !== 'upgraded') {
-            return $documents;
+            return $modified;
         }
 
         try {
@@ -587,9 +598,15 @@ public function createDocuments(
                 $clones[] = $clone;
             }
 
-            $this->destination->setPreserveDates(true);
-            $this->destination->createDocuments($collection, $clones, $batchSize);
-            $this->destination->setPreserveDates(false);
+            $this->destination->withPreserveDates(
+                fn () =>
+                $this->destination->createDocuments(
+                    $collection,
+                    $clones,
+                    $batchSize,
+                    null,
+                )
+            );
 
             foreach ($clones as $clone) {
                 foreach ($this->writeFilters as $filter) {
@@ -601,12 +618,11 @@ public function createDocuments(
                     );
                 }
             }
-
         } catch (\Throwable $err) {
             $this->logError('createDocuments', $err);
         }
 
-        return $documents;
+        return $modified;
     }
 
     public function updateDocument(string $collection, string $id, Document $document): Document
@@ -661,20 +677,32 @@ public function updateDocuments(
         string $collection,
         Document $updates,
         array $queries = [],
-        int $batchSize = self::INSERT_BATCH_SIZE
-    ): array {
-        $documents = $this->source->updateDocuments($collection, $updates, $queries, $batchSize);
+        int $batchSize = self::INSERT_BATCH_SIZE,
+        ?callable $onNext = null,
+    ): int {
+        $modified = 0;
+
+        $this->source->updateDocuments(
+            $collection,
+            $updates,
+            $queries,
+            $batchSize,
+            function ($doc) use ($onNext, &$modified) {
+                $onNext && $onNext($doc);
+                $modified++;
+            }
+        );
 
         if (
             \in_array($collection, self::SOURCE_ONLY_COLLECTIONS)
             || $this->destination === null
         ) {
-            return $documents;
+            return $modified;
         }
 
         $upgrade = $this->silent(fn () => $this->getUpgradeStatus($collection));
         if ($upgrade === null || $upgrade->getAttribute('status', '') !== 'upgraded') {
-            return $documents;
+            return $modified;
         }
 
         try {
@@ -690,9 +718,16 @@ public function updateDocuments(
                 );
             }
 
-            $this->destination->setPreserveDates(true);
-            $this->destination->updateDocuments($collection, $clone, $queries, $batchSize);
-            $this->destination->setPreserveDates(false);
+            $modified = $this->destination->withPreserveDates(
+                fn () =>
+                $this->destination->updateDocuments(
+                    $collection,
+                    $clone,
+                    $queries,
+                    $batchSize,
+                    null,
+                )
+            );
 
             foreach ($this->writeFilters as $filter) {
                 $filter->afterUpdateDocuments(
@@ -707,7 +742,7 @@ public function updateDocuments(
             $this->logError('updateDocuments', $err);
         }
 
-        return $documents;
+        return $modified;
     }
 
     public function deleteDocument(string $collection, string $id): bool
@@ -753,20 +788,34 @@ public function deleteDocument(string $collection, string $id): bool
         return $result;
     }
 
-    public function deleteDocuments(string $collection, array $queries = [], int $batchSize = self::DELETE_BATCH_SIZE): array
-    {
-        $documents = $this->source->deleteDocuments($collection, $queries, $batchSize);
+    public function deleteDocuments(
+        string $collection,
+        array $queries = [],
+        int $batchSize = self::DELETE_BATCH_SIZE,
+        ?callable $onNext = null,
+    ): int {
+        $modified = 0;
+
+        $this->source->deleteDocuments(
+            $collection,
+            $queries,
+            $batchSize,
+            function ($doc) use (&$modified, $onNext) {
+                $onNext && $onNext($doc);
+                $modified++;
+            },
+        );
 
         if (
             \in_array($collection, self::SOURCE_ONLY_COLLECTIONS)
             || $this->destination === null
         ) {
-            return $documents;
+            return $modified;
         }
 
         $upgrade = $this->silent(fn () => $this->getUpgradeStatus($collection));
         if ($upgrade === null || $upgrade->getAttribute('status', '') !== 'upgraded') {
-            return $documents;
+            return $modified;
         }
 
         try {
@@ -779,7 +828,12 @@ public function deleteDocuments(string $collection, array $queries = [], int $ba
                 );
             }
 
-            $this->destination->deleteDocuments($collection, $queries, $batchSize);
+            $modified = $this->destination->deleteDocuments(
+                $collection,
+                $queries,
+                $batchSize,
+                null,
+            );
 
             foreach ($this->writeFilters as $filter) {
                 $filter->afterDeleteDocuments(
@@ -793,7 +847,7 @@ public function deleteDocuments(string $collection, array $queries = [], int $ba
             $this->logError('deleteDocuments', $err);
         }
 
-        return $documents;
+        return $modified;
     }
 
     public function updateAttributeRequired(string $collection, string $id, bool $required): Document
diff --git a/tests/e2e/Adapter/Base.php b/tests/e2e/Adapter/Base.php
index 29829ef25..b42d536a2 100644
--- a/tests/e2e/Adapter/Base.php
+++ b/tests/e2e/Adapter/Base.php
@@ -916,7 +916,7 @@ public function testPreserveDatesCreate(): void
                 'attr1' => 'value3',
                 '$createdAt' => $date
             ]),
-        ], 2);
+        ], batchSize: 2);
 
         $doc1 = static::getDatabase()->getDocument('preserve_create_dates', 'doc1');
         $doc2 = static::getDatabase()->getDocument('preserve_create_dates', 'doc2');
@@ -2284,9 +2284,9 @@ public function testCreateDocuments(): array
             ]);
         }
 
-        $documents = static::getDatabase()->createDocuments($collection, $documents, 3);
+        $count = static::getDatabase()->createDocuments($collection, $documents, 3);
 
-        $this->assertEquals($count, count($documents));
+        $this->assertEquals($count, \count($documents));
 
         foreach ($documents as $document) {
             $this->assertNotEmpty(true, $document->getId());
@@ -2325,16 +2325,19 @@ public function testCreateDocumentsWithDifferentAttributes(): void
             ]),
         ];
 
-        $documents = static::getDatabase()->createDocuments($collection, $documents);
+        $results = [];
+        $count = static::getDatabase()->createDocuments($collection, $documents, onNext: function ($doc) use (&$results) {
+            $results[] = $doc;
+        });
 
-        $this->assertEquals(2, count($documents));
+        $this->assertEquals(2, $count);
 
-        $this->assertEquals('text📝', $documents[0]->getAttribute('string'));
-        $this->assertEquals(5, $documents[0]->getAttribute('integer'));
-        $this->assertEquals('not_default', $documents[0]->getAttribute('string_default'));
-        $this->assertEquals('text📝', $documents[1]->getAttribute('string'));
-        $this->assertNull($documents[1]->getAttribute('integer'));
-        $this->assertEquals('default', $documents[1]->getAttribute('string_default'));
+        $this->assertEquals('text📝', $results[0]->getAttribute('string'));
+        $this->assertEquals(5, $results[0]->getAttribute('integer'));
+        $this->assertEquals('not_default', $results[0]->getAttribute('string_default'));
+        $this->assertEquals('text📝', $results[1]->getAttribute('string'));
+        $this->assertEquals(null, $results[1]->getAttribute('integer'));
+        $this->assertEquals('default', $results[1]->getAttribute('string_default'));
 
         /**
          * Expect fail, mix of internalId and no internalId
@@ -2406,11 +2409,14 @@ public function testCreateOrUpdateDocuments(): void
             ]),
         ];
 
-        $documents = static::getDatabase()->createOrUpdateDocuments(__FUNCTION__, $documents);
+        $results = [];
+        $count = static::getDatabase()->createOrUpdateDocuments(__FUNCTION__, $documents, onNext: function ($doc) use (&$results) {
+            $results[] = $doc;
+        });
 
-        $this->assertEquals(2, count($documents));
+        $this->assertEquals(2, $count);
 
-        foreach ($documents as $document) {
+        foreach ($results as $document) {
             $this->assertNotEmpty(true, $document->getId());
             $this->assertIsString($document->getAttribute('string'));
             $this->assertEquals('text📝', $document->getAttribute('string')); // Also makes sure an emoji is working
@@ -2439,11 +2445,14 @@ public function testCreateOrUpdateDocuments(): void
         $documents[1]->setAttribute('string', 'new text📝');
         $documents[1]->setAttribute('integer', 10);
 
-        $documents = static::getDatabase()->createOrUpdateDocuments(__FUNCTION__, $documents);
+        $results = [];
+        $count = static::getDatabase()->createOrUpdateDocuments(__FUNCTION__, $documents, onNext: function ($doc) use (&$results) {
+            $results[] = $doc;
+        });
 
-        $this->assertEquals(2, count($documents));
+        $this->assertEquals(2, $count);
 
-        foreach ($documents as $document) {
+        foreach ($results as $document) {
             $this->assertNotEmpty(true, $document->getId());
             $this->assertIsString($document->getAttribute('string'));
             $this->assertEquals('new text📝', $document->getAttribute('string')); // Also makes sure an emoji is working
@@ -2576,13 +2585,17 @@ public function testCreateOrUpdateDocumentsPermissions(): void
 
         static::getDatabase()->createOrUpdateDocuments(__FUNCTION__, [$document]);
 
-        $documents = static::getDatabase()->createOrUpdateDocuments(
+        $results = [];
+        $count = static::getDatabase()->createOrUpdateDocuments(
             __FUNCTION__,
-            [$document->setAttribute('string', 'updated')]
+            [$document->setAttribute('string', 'updated')],
+            onNext: function ($doc) use (&$results) {
+                $results[] = $doc;
+            }
         );
 
-        $this->assertEquals(1, count($documents));
-        $this->assertEquals('updated', $documents[0]->getAttribute('string'));
+        $this->assertEquals(1, $count);
+        $this->assertEquals('updated', $results[0]->getAttribute('string'));
 
         $document = new Document([
             '$id' => 'third',
@@ -2602,13 +2615,17 @@ public function testCreateOrUpdateDocumentsPermissions(): void
             Permission::delete(Role::user('user1')),
         ];
 
-        $documents = static::getDatabase()->createOrUpdateDocuments(
+        $results = [];
+        $count = static::getDatabase()->createOrUpdateDocuments(
             __FUNCTION__,
-            [$document->setAttribute('$permissions', $newPermissions)]
+            [$document->setAttribute('$permissions', $newPermissions)],
+            onNext: function ($doc) use (&$results) {
+                $results[] = $doc;
+            }
         );
 
-        $this->assertEquals(1, count($documents));
-        $this->assertEquals($newPermissions, $documents[0]->getPermissions());
+        $this->assertEquals(1, $count);
+        $this->assertEquals($newPermissions, $results[0]->getPermissions());
 
         $document = static::getDatabase()->getDocument(__FUNCTION__, 'third');
 
@@ -16783,9 +16800,9 @@ public function testDeleteBulkDocuments(): void
          */
         $selects = ['$internalId', '$id', '$collection', '$permissions', '$updatedAt'];
 
-        $this->assertCount(2, static::getDatabase()->deleteDocuments(
-            'bulk_delete',
-            [
+        $count = static::getDatabase()->deleteDocuments(
+            collection: 'bulk_delete',
+            queries: [
                 Query::select([...$selects, '$createdAt']),
                 Query::cursorAfter($docs[6]),
                 Query::greaterThan('$createdAt', '2000-01-01'),
@@ -16793,11 +16810,13 @@ public function testDeleteBulkDocuments(): void
                 Query::orderAsc(),
                 Query::limit(2),
             ],
-            1
-        ));
+            batchSize: 1
+        );
+
+        $this->assertEquals(2, $count);
 
         // TEST: Bulk Delete All Documents
-        $this->assertCount(8, static::getDatabase()->deleteDocuments('bulk_delete'));
+        $this->assertEquals(8, static::getDatabase()->deleteDocuments('bulk_delete'));
 
         $docs = static::getDatabase()->find('bulk_delete');
         $this->assertCount(0, $docs);
@@ -16805,17 +16824,21 @@ public function testDeleteBulkDocuments(): void
         // TEST: Bulk delete documents with queries.
         $this->propagateBulkDocuments('bulk_delete');
 
-        $modified = static::getDatabase()->deleteDocuments('bulk_delete', [
+        $results = [];
+        $count = static::getDatabase()->deleteDocuments('bulk_delete', [
             Query::greaterThanEqual('integer', 5)
-        ]);
-        $this->assertCount(5, $modified);
+        ], onNext: function ($doc) use (&$results) {
+            $results[] = $doc;
+        });
+
+        $this->assertEquals(5, $count);
 
-        foreach ($modified as $document) {
+        foreach ($results as $document) {
             $this->assertGreaterThanOrEqual(5, $document->getAttribute('integer'));
         }
 
         $docs = static::getDatabase()->find('bulk_delete');
-        $this->assertCount(5, $docs);
+        $this->assertEquals(5, \count($docs));
 
         // TEST (FAIL): Can't delete documents in the past
         $oneHourAgo = (new \DateTime())->sub(new \DateInterval('PT1H'));
@@ -16843,8 +16866,8 @@ public function testDeleteBulkDocuments(): void
             Permission::delete(Role::any())
         ], false);
 
-        $this->assertCount(5, static::getDatabase()->deleteDocuments('bulk_delete'));
-        $this->assertEquals(0, count($this->getDatabase()->find('bulk_delete')));
+        $this->assertEquals(5, static::getDatabase()->deleteDocuments('bulk_delete'));
+        $this->assertEquals(0, \count($this->getDatabase()->find('bulk_delete')));
 
         // TEST: Make sure we can't delete documents we don't have permissions for
         static::getDatabase()->updateCollection('bulk_delete', [
@@ -16852,21 +16875,23 @@ public function testDeleteBulkDocuments(): void
         ], true);
         $this->propagateBulkDocuments('bulk_delete', documentSecurity: true);
 
-        $this->assertCount(0, static::getDatabase()->deleteDocuments('bulk_delete'));
+        $this->assertEquals(0, static::getDatabase()->deleteDocuments('bulk_delete'));
 
         $documents = Authorization::skip(function () {
             return static::getDatabase()->find('bulk_delete');
         });
 
-        $this->assertCount(10, $documents);
+        $this->assertEquals(10, \count($documents));
 
         static::getDatabase()->updateCollection('bulk_delete', [
             Permission::create(Role::any()),
             Permission::read(Role::any()),
             Permission::delete(Role::any())
         ], false);
+
         static::getDatabase()->deleteDocuments('bulk_delete');
-        $this->assertEquals(0, count($this->getDatabase()->find('bulk_delete')));
+
+        $this->assertEquals(0, \count($this->getDatabase()->find('bulk_delete')));
 
         // Teardown
         static::getDatabase()->deleteCollection('bulk_delete');
@@ -16906,30 +16931,30 @@ public function testDeleteBulkDocumentsQueries(): void
         // Test limit
         $this->propagateBulkDocuments('bulk_delete_queries');
 
-        $this->assertCount(5, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::limit(5)]));
-        $this->assertCount(5, static::getDatabase()->find('bulk_delete_queries'));
+        $this->assertEquals(5, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::limit(5)]));
+        $this->assertEquals(5, \count(static::getDatabase()->find('bulk_delete_queries')));
 
-        $this->assertCount(5, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::limit(5)]));
-        $this->assertCount(0, static::getDatabase()->find('bulk_delete_queries'));
+        $this->assertEquals(5, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::limit(5)]));
+        $this->assertEquals(0, \count(static::getDatabase()->find('bulk_delete_queries')));
 
         // Test Limit more than batchSize
         $this->propagateBulkDocuments('bulk_delete_queries', Database::DELETE_BATCH_SIZE * 2);
-        $this->assertCount(Database::DELETE_BATCH_SIZE * 2, static::getDatabase()->find('bulk_delete_queries', [Query::limit(Database::DELETE_BATCH_SIZE * 2)]));
-        $this->assertCount(Database::DELETE_BATCH_SIZE + 2, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::limit(Database::DELETE_BATCH_SIZE + 2)]));
-        $this->assertCount(Database::DELETE_BATCH_SIZE - 2, static::getDatabase()->find('bulk_delete_queries', [Query::limit(Database::DELETE_BATCH_SIZE * 2)]));
-        $this->assertCount(Database::DELETE_BATCH_SIZE - 2, $this->getDatabase()->deleteDocuments('bulk_delete_queries'));
+        $this->assertEquals(Database::DELETE_BATCH_SIZE * 2, \count(static::getDatabase()->find('bulk_delete_queries', [Query::limit(Database::DELETE_BATCH_SIZE * 2)])));
+        $this->assertEquals(Database::DELETE_BATCH_SIZE + 2, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::limit(Database::DELETE_BATCH_SIZE + 2)]));
+        $this->assertEquals(Database::DELETE_BATCH_SIZE - 2, \count(static::getDatabase()->find('bulk_delete_queries', [Query::limit(Database::DELETE_BATCH_SIZE * 2)])));
+        $this->assertEquals(Database::DELETE_BATCH_SIZE - 2, $this->getDatabase()->deleteDocuments('bulk_delete_queries'));
 
         // Test Offset
         $this->propagateBulkDocuments('bulk_delete_queries', 100);
-        $this->assertCount(50, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::offset(50)]));
+        $this->assertEquals(50, static::getDatabase()->deleteDocuments('bulk_delete_queries', [Query::offset(50)]));
 
         $docs = static::getDatabase()->find('bulk_delete_queries', [Query::limit(100)]);
-        $this->assertCount(50, $docs);
+        $this->assertEquals(50, \count($docs));
 
-        $lastDoc = end($docs);
+        $lastDoc = \end($docs);
         $this->assertNotEmpty($lastDoc);
         $this->assertEquals('doc49', $lastDoc->getId());
-        $this->assertCount(50, static::getDatabase()->deleteDocuments('bulk_delete_queries'));
+        $this->assertEquals(50, static::getDatabase()->deleteDocuments('bulk_delete_queries'));
 
         static::getDatabase()->deleteCollection('bulk_delete_queries');
     }
@@ -17515,14 +17540,18 @@ public function testUpdateDocuments(): void
         }
 
         // Test Update half of the documents
-        $modified = static::getDatabase()->updateDocuments($collection, new Document([
+        $results = [];
+        $count = static::getDatabase()->updateDocuments($collection, new Document([
             'string' => 'text📝 updated',
         ]), [
             Query::greaterThanEqual('integer', 5),
-        ]);
-        $this->assertCount(5, $modified);
+        ], onNext: function ($doc) use (&$results) {
+            $results[] = $doc;
+        });
 
-        foreach ($modified as $document) {
+        $this->assertEquals(5, $count);
+
+        foreach ($results as $document) {
             $this->assertEquals('text📝 updated', $document->getAttribute('string'));
         }
 
@@ -17548,7 +17577,7 @@ public function testUpdateDocuments(): void
         }
 
         // Test Update all documents
-        $this->assertCount(10, static::getDatabase()->updateDocuments($collection, new Document([
+        $this->assertEquals(10, static::getDatabase()->updateDocuments($collection, new Document([
             'string' => 'text📝 updated all',
         ])));
 
@@ -17565,7 +17594,7 @@ public function testUpdateDocuments(): void
 
         try {
             $this->getDatabase()->withRequestTimestamp($oneHourAgo, function () use ($collection) {
-                return static::getDatabase()->updateDocuments($collection, new Document([
+                static::getDatabase()->updateDocuments($collection, new Document([
                     'string' => 'text📝 updated all',
                 ]));
             });
@@ -17638,7 +17667,7 @@ public function testUpdateDocuments(): void
         });
 
         // Test we can update more documents than batchSize
-        $this->assertCount(10, static::getDatabase()->updateDocuments($collection, new Document([
+        $this->assertEquals(10, static::getDatabase()->updateDocuments($collection, new Document([
             'string' => 'batchSize Test'
         ]), batchSize: 2));
 
@@ -17684,18 +17713,21 @@ public function testUpdateDocumentsQueries(): void
         // Test limit
         $this->propagateBulkDocuments($collection, 100);
 
-        $this->assertCount(10, static::getDatabase()->updateDocuments($collection, new Document([
+        $this->assertEquals(10, static::getDatabase()->updateDocuments($collection, new Document([
             'text' => 'text📝 updated',
         ]), [Query::limit(10)]));
-        $this->assertCount(10, static::getDatabase()->find($collection, [Query::equal('text', ['text📝 updated'])]));
-        $this->assertCount(100, static::getDatabase()->deleteDocuments($collection));
-        $this->assertCount(0, static::getDatabase()->find($collection));
+
+        $this->assertEquals(10, \count(static::getDatabase()->find($collection, [Query::equal('text', ['text📝 updated'])])));
+        $this->assertEquals(100, static::getDatabase()->deleteDocuments($collection));
+        $this->assertEquals(0, \count(static::getDatabase()->find($collection)));
 
         // Test Offset
         $this->propagateBulkDocuments($collection, 100);
-        $this->assertCount(50, static::getDatabase()->updateDocuments($collection, new Document([
+        $this->assertEquals(50, static::getDatabase()->updateDocuments($collection, new Document([
             'text' => 'text📝 updated',
-        ]), [Query::offset(50)]));
+        ]), [
+            Query::offset(50),
+        ]));
 
         $docs = static::getDatabase()->find($collection, [Query::equal('text', ['text📝 updated']), Query::limit(100)]);
         $this->assertCount(50, $docs);
@@ -17704,7 +17736,7 @@ public function testUpdateDocumentsQueries(): void
         $this->assertNotEmpty($lastDoc);
         $this->assertEquals('doc99', $lastDoc->getId());
 
-        $this->assertCount(100, static::getDatabase()->deleteDocuments($collection));
+        $this->assertEquals(100, static::getDatabase()->deleteDocuments($collection));
     }
 
     public function testUpdateDocumentsPermissions(): void
@@ -17752,7 +17784,7 @@ public function testUpdateDocumentsPermissions(): void
             ]));
         });
 
-        $affected = static::getDatabase()->updateDocuments($collection, new Document([
+        $modified = static::getDatabase()->updateDocuments($collection, new Document([
             '$permissions' => [
                 Permission::read(Role::user('user2')),
                 Permission::create(Role::user('user2')),
@@ -17765,8 +17797,8 @@ public function testUpdateDocumentsPermissions(): void
             return static::getDatabase()->find($collection);
         });
 
-        $this->assertCount(10, $affected);
-        $this->assertCount(11, $documents);
+        $this->assertEquals(10, $modified);
+        $this->assertEquals(11, \count($documents));
 
         $modifiedDocuments = array_filter($documents, function (Document $document) {
             return $document->getAttribute('$permissions') == [
@@ -17793,7 +17825,7 @@ public function testUpdateDocumentsPermissions(): void
         Authorization::setRole(Role::user('user2')->toString());
 
         // Test Bulk permission update with data
-        $affected = static::getDatabase()->updateDocuments($collection, new Document([
+        $modified = static::getDatabase()->updateDocuments($collection, new Document([
             '$permissions' => [
                 Permission::read(Role::user('user3')),
                 Permission::create(Role::user('user3')),
@@ -17803,7 +17835,7 @@ public function testUpdateDocumentsPermissions(): void
             'string' => 'text📝 updated',
         ]));
 
-        $this->assertCount(10, $affected);
+        $this->assertEquals(10, $modified);
 
         $documents = Authorization::skip(function () use ($collection) {
             return $this->getDatabase()->find($collection);