Repository: appwrite/appwrite
File: src/Appwrite/Platform/Workers/Migrations.php (PHP)

Code before:
            $migration->setAttribute('status', 'processing');
            $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

            $source = $this->processSource($migration);
            $destination = $this->processDestination($migration, $tempAPIKey);

            $transfer = new Transfer(
                $source,
                $destination
            );

            $aggregatedResources = [];
            /** Start Transfer */
            if (empty($source->getErrors())) {
                $migration->setAttribute('stage', 'migrating');
                $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

                $transfer->run(
                    $migration->getAttribute('resources'),
                    function ($resources) use ($migration, $transfer, $projectDocument, $queueForRealtime, &$aggregatedResources) {
                        $migration->setAttribute('resourceData', json_encode($transfer->getCache()));
                        $migration->setAttribute('statusCounters', json_encode($transfer->getStatusCounters()));

                        if (!empty($resources)) {
                            $aggregatedResources[] = $resources;
                        }
                        $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);
                    },
                    $migration->getAttribute('resourceId'),
                    $migration->getAttribute('resourceType')
                );
            }

            $destination->shutDown();
            $source->shutDown();

            $sourceErrors = $source->getErrors();
            $destinationErrors = $destination->getErrors();

            if (! empty($sourceErrors) || ! empty($destinationErrors)) {
                $migration->setAttribute('status', 'failed');
                $migration->setAttribute('stage', 'finished');

                $errorMessages = [];
                foreach ($sourceErrors as $error) {
                    $errorMessages[] = json_encode($error);
                }
                foreach ($destinationErrors as $error) {
                    $errorMessages[] = json_encode($error);
                }


Review comment (bug):
This could get very heavy on RAM if there are millions of docs, let's buffer and flush 1000 at a time. Doesn't actually matter if the migrations fails later - the docs that _were_ created are still in the DB

Code after:
            $migration->setAttribute('status', 'processing');
            $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

            $source = $this->processSource($migration);
            $destination = $this->processDestination($migration, $tempAPIKey);

            $transfer = new Transfer(
                $source,
                $destination
            );

            $aggregatedResources = [];
            /** Start Transfer */
            if (empty($source->getErrors())) {
                $migration->setAttribute('stage', 'migrating');
                $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

                $transfer->run(
                    $migration->getAttribute('resources'),
                    function ($resources) use ($migration, $transfer, $projectDocument, $queueForRealtime, &$aggregatedResources) {
                        $migration->setAttribute('resourceData', json_encode($transfer->getCache()));
                        $migration->setAttribute('statusCounters', json_encode($transfer->getStatusCounters()));

                        if (!empty($resources)) {
                            /**
                             * @var Resource $resource
                            */
                            $resource = $resources[0];
                            $count = count($resources);
                            $databaseId = null;
                            $tableId = null;
                            switch ($resource->getName()) {
                                case ResourceTable::getName():
                                    /** @var ResourceTable $resource */
                                    $databaseId = $resource->getDatabase()->getSequence();
                                    break;
                                case ResourceRow::getName():
                                    /** @var ResourceRow $resource */
                                    $table = $resource->getTable();
                                    $databaseId = $table->getDatabase()->getSequence();
                                    $tableId = $table->getSequence();
                                    break;
                                default:
                                    break;
                            }
                            $aggregatedResources[] = [
                                'name' => $resource->getName(),
                                'count' => $count,
                                'databaseId' => $databaseId,
                                'tableId' => $tableId
                            ];