Appwrite.php

<?php

namespace Utopia\Migration\Sources;

use Appwrite\AppwriteException;
use Appwrite\Client;
use Appwrite\Query;
use Appwrite\Services\Databases;
use Appwrite\Services\Functions;
use Appwrite\Services\Messaging;
use Appwrite\Services\Sites;
use Appwrite\Services\Storage;
use Appwrite\Services\Teams;
use Appwrite\Services\Users;
use Utopia\Database\Database as UtopiaDatabase;
use Utopia\Database\DateTime as UtopiaDateTime;
use Utopia\Database\Document as UtopiaDocument;
use Utopia\Migration\Exception;
use Utopia\Migration\Resource;
use Utopia\Migration\Resources\Auth\Hash;
use Utopia\Migration\Resources\Auth\Membership;
use Utopia\Migration\Resources\Auth\Team;
use Utopia\Migration\Resources\Auth\User;
use Utopia\Migration\Resources\Database\Attribute;
use Utopia\Migration\Resources\Database\Collection;
use Utopia\Migration\Resources\Database\Column;
use Utopia\Migration\Resources\Database\Columns\Boolean;
use Utopia\Migration\Resources\Database\Columns\DateTime;
use Utopia\Migration\Resources\Database\Columns\Decimal;
use Utopia\Migration\Resources\Database\Columns\Email;
use Utopia\Migration\Resources\Database\Columns\Enum;
use Utopia\Migration\Resources\Database\Columns\Integer;
use Utopia\Migration\Resources\Database\Columns\IP;
use Utopia\Migration\Resources\Database\Columns\Line;
use Utopia\Migration\Resources\Database\Columns\LongText;
use Utopia\Migration\Resources\Database\Columns\MediumText;
use Utopia\Migration\Resources\Database\Columns\ObjectType;
use Utopia\Migration\Resources\Database\Columns\Point;
use Utopia\Migration\Resources\Database\Columns\Polygon;
use Utopia\Migration\Resources\Database\Columns\RegularText;
use Utopia\Migration\Resources\Database\Columns\Relationship;
use Utopia\Migration\Resources\Database\Columns\Text;
use Utopia\Migration\Resources\Database\Columns\URL;
use Utopia\Migration\Resources\Database\Columns\Varchar;
use Utopia\Migration\Resources\Database\Columns\Vector;
use Utopia\Migration\Resources\Database\Database;
use Utopia\Migration\Resources\Database\Document;
use Utopia\Migration\Resources\Database\DocumentsDB;
use Utopia\Migration\Resources\Database\Index;
use Utopia\Migration\Resources\Database\Row;
use Utopia\Migration\Resources\Database\Table;
use Utopia\Migration\Resources\Database\VectorsDB;
use Utopia\Migration\Resources\Functions\Deployment;
use Utopia\Migration\Resources\Functions\EnvVar;
use Utopia\Migration\Resources\Functions\Func;
use Utopia\Migration\Resources\Messaging\Message;
use Utopia\Migration\Resources\Messaging\Provider;
use Utopia\Migration\Resources\Messaging\Subscriber;
use Utopia\Migration\Resources\Messaging\Topic;
use Utopia\Migration\Resources\Sites\Deployment as SiteDeployment;
use Utopia\Migration\Resources\Sites\EnvVar as SiteEnvVar;
use Utopia\Migration\Resources\Sites\Site;
use Utopia\Migration\Resources\Storage\Bucket;
use Utopia\Migration\Resources\Storage\File;
use Utopia\Migration\Source;
use Utopia\Migration\Sources\Appwrite\Reader;
use Utopia\Migration\Sources\Appwrite\Reader\API as APIReader;
use Utopia\Migration\Sources\Appwrite\Reader\Database as DatabaseReader;
use Utopia\Migration\Transfer;

class Appwrite extends Source
{
    public const SOURCE_API = 'api';
    public const SOURCE_DATABASE = 'database';

    private const DEFAULT_PAGE_LIMIT = 25;

    protected Client $client;

    private Reader $reader;

    private Users $users;

    private Teams $teams;

    private Storage $storage;

    private Functions $functions;

    private Messaging $messaging;

    private Sites $sites;

    /**
     * @var callable(UtopiaDocument $database|null): UtopiaDatabase
     */
    protected mixed $getDatabasesDB;

    /**
     * @throws \Exception
     */
    public function __construct(
        protected string $project,
        protected string $endpoint,
        protected string $key,
        callable $getDatabasesDB,
        protected string $source = self::SOURCE_API,
        protected ?UtopiaDatabase $dbForProject = null,
        protected array $queries = [],
    ) {
        $this->client = (new Client())
            ->setEndpoint($endpoint)
            ->setProject($project)
            ->setKey($key);

        $this->users = new Users($this->client);
        $this->teams = new Teams($this->client);
        $this->storage = new Storage($this->client);
        $this->functions = new Functions($this->client);
        $this->messaging = new Messaging($this->client);
        $this->sites = new Sites($this->client);

        $this->headers['x-appwrite-project'] = $this->project;
        $this->headers['x-appwrite-key'] = $this->key;

        $this->getDatabasesDB = $getDatabasesDB;

        switch ($this->source) {
            case static::SOURCE_API:
                $this->reader = new APIReader(new Databases($this->client));
                break;

            case static::SOURCE_DATABASE:
                if (\is_null($dbForProject)) {
                    throw new \Exception('Database is required for database source', Exception::CODE_VALIDATION);
                }
                $this->reader = new DatabaseReader(
                    $this->dbForProject,
                    $this->getDatabasesDB,
                    $this->project
                );
                break;

            default:
                throw new \Exception('Unknown source', Exception::CODE_VALIDATION);
        }

    }

    public static function getName(): string
    {
        return 'Appwrite';
    }

    /**
     * @return array<string>
     */
    public static function getSupportedResources(): array
    {
        return [
            // Auth
            Resource::TYPE_USER,
            Resource::TYPE_TEAM,
            Resource::TYPE_MEMBERSHIP,

            // Database
            Resource::TYPE_DATABASE,
            Resource::TYPE_TABLE,
            Resource::TYPE_COLUMN,
            Resource::TYPE_INDEX,
            Resource::TYPE_ROW,

            // legacy
            Resource::TYPE_DOCUMENT,
            Resource::TYPE_ATTRIBUTE,
            Resource::TYPE_COLLECTION,

            // documentsdb
            Resource::TYPE_DATABASE_DOCUMENTSDB,
            // vectorsdb
            Resource::TYPE_DATABASE_VECTORSDB,

            // Storage
            Resource::TYPE_BUCKET,
            Resource::TYPE_FILE,

            // Functions
            Resource::TYPE_FUNCTION,
            Resource::TYPE_DEPLOYMENT,
            Resource::TYPE_ENVIRONMENT_VARIABLE,

            // Messaging
            Resource::TYPE_PROVIDER,
            Resource::TYPE_TOPIC,
            Resource::TYPE_SUBSCRIBER,
            Resource::TYPE_MESSAGE,

            // Sites
            Resource::TYPE_SITE,
            Resource::TYPE_SITE_DEPLOYMENT,
            Resource::TYPE_SITE_VARIABLE,

            // Settings
        ];
    }

    /**
     * @return int
     */
    public function getDatabasesBatchSize(): int
    {
        return match ($this->source) {
            static::SOURCE_API => 500,
            static::SOURCE_DATABASE => 1000,
        };
    }

    /**
     * @param array<string> $resources
     * @param array<string, array<string>> $resourceIds
     * @return array<string, mixed>
     *
     * @throws \Exception
     */
    public function report(array $resources = [], array $resourceIds = []): array
    {
        $this->validateResourceIds($resourceIds);

        $report = [];

        if (empty($resources)) {
            $resources = $this->getSupportedResources();
        }

        try {
            $this->reportAuth($resources, $report, $resourceIds);
            $this->reportDatabases($resources, $report, $resourceIds);
            $this->reportStorage($resources, $report, $resourceIds);
            $this->reportFunctions($resources, $report, $resourceIds);
            $this->reportMessaging($resources, $report, $resourceIds);
            $this->reportSites($resources, $report, $resourceIds);

            $report['version'] = $this->call(
                'GET',
                '/health/version',
                [
                    'x-appwrite-key' => '',
                    'x-appwrite-project' => '',
                ]
            )['version'];
        } catch (\Throwable $e) {
            if ($e->getCode() === 403) {
                throw new \Exception('Missing required scopes.', $e->getCode(), $e);
            } else {
                throw new \Exception($e->getMessage(), $e->getCode(), $e);
            }
        }

        $this->previousReport = $report;

        return $report;
    }

    /**
     * @param array $resources
     * @param array $report
     * @param array<string, array<string>> $resourceIds
     * @throws AppwriteException
     */
    private function reportAuth(array $resources, array &$report, array $resourceIds = []): void
    {
        // check if we need to fetch teams!
        $needTeams = !empty(array_intersect(
            [Resource::TYPE_TEAM, Resource::TYPE_MEMBERSHIP],
            $resources
        ));

        $teams = ['total' => 0, 'teams' => []];

        if (\in_array(Resource::TYPE_USER, $resources)) {
            $userQueries = $this->buildQueries(
                resourceType: Resource::TYPE_USER,
                resourceIds: $resourceIds,
                limit: 1
            );
            $userList = $this->users->list($userQueries);
            $report[Resource::TYPE_USER] = $userList['total'];
        }

        if ($needTeams) {
            if (\in_array(Resource::TYPE_MEMBERSHIP, $resources)) {
                $allTeams = [];
                $lastTeam = null;

                while (true) {
                    $params = $this->buildQueries(
                        resourceType: Resource::TYPE_TEAM,
                        resourceIds: $resourceIds,
                        cursor: $lastTeam
                    );
                    $teamList = $this->teams->list($params);

                    $totalTeams = $teamList['total'];
                    $currentTeams = $teamList['teams'];

                    $allTeams = array_merge($allTeams, $currentTeams);
                    $lastTeam = $currentTeams[count($currentTeams) - 1]['$id'] ?? null;

                    if (count($currentTeams) < self::DEFAULT_PAGE_LIMIT) {
                        break;
                    }
                }
                $teams = ['total' => $totalTeams, 'teams' => $allTeams];
            } else {
                $params = $this->buildQueries(
                    resourceType: Resource::TYPE_TEAM,
                    resourceIds: $resourceIds,
                    limit: 1
                );
                $teamList = $this->teams->list($params);
                $teams = ['total' => $teamList['total'], 'teams' => []];
            }
        }

        if (\in_array(Resource::TYPE_TEAM, $resources)) {
            $report[Resource::TYPE_TEAM] = $teams['total'];
        }

        if (\in_array(Resource::TYPE_MEMBERSHIP, $resources)) {
            $report[Resource::TYPE_MEMBERSHIP] = 0;
            foreach ($teams['teams'] as $team) {
                $report[Resource::TYPE_MEMBERSHIP] += $this->teams->listMemberships(
                    $team['$id'],
                    [Query::limit(1)]
                )['total'];
            }
        }
    }

    /**
     * @throws Exception
     * @throws AppwriteException
     */
    private function reportDatabases(array $resources, array &$report, array $resourceIds = []): void
    {
        $this->reader->report($resources, $report, $resourceIds);
    }

    /**
     * @param array $resources
     * @param array $report
     * @param array<string, array<string>> $resourceIds
     * @throws AppwriteException
     */
    private function reportStorage(array $resources, array &$report, array $resourceIds = []): void
    {

        if (\in_array(Resource::TYPE_BUCKET, $resources)) {
            $bucketQueries = $this->buildQueries(
                resourceType: Resource::TYPE_BUCKET,
                resourceIds: $resourceIds,
                limit: 1
            );
            $report[Resource::TYPE_BUCKET] = $this->storage->listBuckets($bucketQueries)['total'];
        }

        if (\in_array(Resource::TYPE_FILE, $resources)) {
            $report[Resource::TYPE_FILE] = 0;
            $report['size'] = 0;
            $buckets = [];
            $lastBucket = null;

            while (true) {
                $queries = $this->buildQueries(
                    resourceType: Resource::TYPE_BUCKET,
                    resourceIds: $resourceIds,
                    cursor: $lastBucket,
                );
                $currentBuckets = $this->storage->listBuckets($queries)['buckets'];

                $buckets = array_merge($buckets, $currentBuckets);
                $lastBucket = $buckets[count($buckets) - 1]['$id'] ?? null;

                if (count($currentBuckets) < self::DEFAULT_PAGE_LIMIT) {
                    break;
                }
            }

            foreach ($buckets as $bucket) {
                $filesResponse = $this->storage->listFiles(
                    $bucket['$id'],
                    [Query::limit(1)]
                );

                $report['size'] += $bucket['totalSize'] ?? 0;
                $report[Resource::TYPE_FILE] += $filesResponse['total'];
            }

            $report['size'] = $report['size'] / 1000 / 1000; // MB
        }
    }

    private function reportFunctions(array $resources, array &$report, array $resourceIds = []): void
    {
        $needVarsOrDeployments = (
            \in_array(Resource::TYPE_DEPLOYMENT, $resources) ||
            \in_array(Resource::TYPE_ENVIRONMENT_VARIABLE, $resources)
        );

        $functions = [];
        $totalFunctions = 0;

        if (!$needVarsOrDeployments && \in_array(Resource::TYPE_FUNCTION, $resources)) {
            $functionQueries = $this->buildQueries(
                resourceType: Resource::TYPE_FUNCTION,
                resourceIds: $resourceIds,
                limit: 1
            );
            $report[Resource::TYPE_FUNCTION] = $this->functions->list($functionQueries)['total'];
            return;
        }

        if ($needVarsOrDeployments) {
            $lastFunction = null;
            while (true) {
                $params = $this->buildQueries(
                    resourceType: Resource::TYPE_FUNCTION,
                    resourceIds: $resourceIds,
                    cursor: $lastFunction,
                );
                $funcList = $this->functions->list($params);

                $totalFunctions = $funcList['total'];
                $currentFunctions = $funcList['functions'];
                $functions = array_merge($functions, $currentFunctions);

                $lastFunction = $currentFunctions[count($currentFunctions) - 1]['$id'] ?? null;
                if (count($currentFunctions) < self::DEFAULT_PAGE_LIMIT) {
                    break;
                }
            }
        }

        if (\in_array(Resource::TYPE_FUNCTION, $resources)) {
            $report[Resource::TYPE_FUNCTION] = $totalFunctions;
        }

        if (\in_array(Resource::TYPE_DEPLOYMENT, $resources)) {
            $report[Resource::TYPE_DEPLOYMENT] = 0;
            foreach ($functions as $function) {
                if (!empty($function['deploymentId'])) {
                    $report[Resource::TYPE_DEPLOYMENT] += 1;
                }
            }
        }

        if (\in_array(Resource::TYPE_ENVIRONMENT_VARIABLE, $resources)) {
            $report[Resource::TYPE_ENVIRONMENT_VARIABLE] = 0;
            foreach ($functions as $function) {
                // function model contains `vars`, we don't need to fetch the list again.
                $report[Resource::TYPE_ENVIRONMENT_VARIABLE] += count($function['vars'] ?? []);
            }
        }
    }

    private function reportSites(array $resources, array &$report, array $resourceIds = []): void
    {
        $needVarsOrDeployments = (
            \in_array(Resource::TYPE_SITE_DEPLOYMENT, $resources) ||
            \in_array(Resource::TYPE_SITE_VARIABLE, $resources)
        );

        $sites = [];
        $totalSites = 0;

        if (!$needVarsOrDeployments && \in_array(Resource::TYPE_SITE, $resources)) {
            $siteQueries = $this->buildQueries(
                resourceType: Resource::TYPE_SITE,
                resourceIds: $resourceIds,
                limit: 1
            );
            $report[Resource::TYPE_SITE] = $this->sites->list($siteQueries)['total'];
            return;
        }

        if ($needVarsOrDeployments) {
            $lastSite = null;
            while (true) {
                $params = $this->buildQueries(
                    resourceType: Resource::TYPE_SITE,
                    resourceIds: $resourceIds,
                    cursor: $lastSite,
                );
                $siteList = $this->sites->list($params);

                $totalSites = $siteList['total'];
                $currentSites = $siteList['sites'];
                $sites = array_merge($sites, $currentSites);

                if (count($currentSites) === 0 || count($currentSites) < self::DEFAULT_PAGE_LIMIT) {
                    break;
                }

                $lastSite = $currentSites[count($currentSites) - 1]['$id'];
            }
        }

        if (\in_array(Resource::TYPE_SITE, $resources)) {
            $report[Resource::TYPE_SITE] = $totalSites;
        }

        if (\in_array(Resource::TYPE_SITE_DEPLOYMENT, $resources)) {
            $report[Resource::TYPE_SITE_DEPLOYMENT] = 0;
            foreach ($sites as $site) {
                if (!empty($site['deploymentId'])) {
                    $report[Resource::TYPE_SITE_DEPLOYMENT] += 1;
                }
            }
        }

        if (\in_array(Resource::TYPE_SITE_VARIABLE, $resources)) {
            $report[Resource::TYPE_SITE_VARIABLE] = 0;
            foreach ($sites as $site) {
                $variables = $this->sites->listVariables($site['$id']);
                $report[Resource::TYPE_SITE_VARIABLE] += $variables['total'] ?? 0;
            }
        }
    }

    /**
     * Export Auth Resources
     *
     * @param int $batchSize Max 100
     * @param array<string> $resources
     */
    protected function exportGroupAuth(int $batchSize, array $resources): void
    {
        try {
            if (\in_array(Resource::TYPE_USER, $resources)) {
                $this->exportUsers($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_USER,
                Transfer::GROUP_AUTH,
                message: $e->getMessage(),
                code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                previous: $e
            ));
        }

        try {
            if (\in_array(Resource::TYPE_TEAM, $resources)) {
                $this->exportTeams($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_TEAM,
                Transfer::GROUP_AUTH,
                message: $e->getMessage(),
                code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                previous: $e
            ));
        }

        try {
            if (\in_array(Resource::TYPE_MEMBERSHIP, $resources)) {
                $this->exportMemberships($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_MEMBERSHIP,
                Transfer::GROUP_AUTH,
                message: $e->getMessage(),
                code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                previous: $e
            ));
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportUsers(int $batchSize): void
    {
        $lastDocument = null;

        while (true) {
            $users = [];

            $queries = [Query::limit($batchSize)];

            if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_USER) {
                $queries[] = Query::equal('$id', $this->rootResourceId);
                $queries[] = Query::limit(1);
            }

            if ($lastDocument) {
                $queries[] = Query::cursorAfter($lastDocument);
            }

            $response = $this->users->list($queries);
            if ($response['total'] == 0) {
                break;
            }

            foreach ($response['users'] as $user) {
                $users[] = new User(
                    $user['$id'],
                    empty($user['email']) ? null : $user['email'],
                    empty($user['name']) ? null : $user['name'],
                    $user['password'] ? new Hash($user['password'], algorithm: $user['hash']) : null,
                    empty($user['phone']) ? null : $user['phone'],
                    $user['labels'] ?? [],
                    '',
                    $user['emailVerification'] ?? false,
                    $user['phoneVerification'] ?? false,
                    !$user['status'],
                    $user['prefs'] ?? [],
                    $user['targets'] ?? [],
                );

                $lastDocument = $user['$id'];
            }

            $this->callback($users);

            if (count($users) < $batchSize) {
                break;
            }
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportTeams(int $batchSize): void
    {
        $this->teams = new Teams($this->client);
        $lastDocument = null;

        while (true) {
            $teams = [];

            $queries = [Query::limit($batchSize)];

            if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_TEAM) {
                $queries[] = Query::equal('$id', $this->rootResourceId);
                $queries[] = Query::limit(1);
            }

            if ($lastDocument) {
                $queries[] = Query::cursorAfter($lastDocument);
            }

            $response = $this->teams->list($queries);
            if ($response['total'] == 0) {
                break;
            }

            foreach ($response['teams'] as $team) {
                $teams[] = new Team(
                    $team['$id'],
                    $team['name'],
                    $team['prefs'],
                );

                $lastDocument = $team['$id'];
            }

            $this->callback($teams);

            if (count($teams) < $batchSize) {
                break;
            }
        }
    }

    /**
     * @throws AppwriteException
     * @throws \Exception
     */
    private function exportMemberships(int $batchSize): void
    {
        $cacheTeams = $this->cache->get(Team::getName());

        /** @var array<string, User> $cacheUsers */
        $cacheUsers = [];

        foreach ($this->cache->get(User::getName()) as $cacheUser) {
            $cacheUsers[$cacheUser->getId()] = $cacheUser;
        }

        foreach ($cacheTeams as $team) {
            /** @var Team $team */
            $lastDocument = null;

            while (true) {
                $memberships = [];

                $queries = [Query::limit($batchSize)];

                if ($lastDocument) {
                    $queries[] = Query::cursorAfter($lastDocument);
                }

                $response = $this->teams->listMemberships($team->getId(), $queries);

                if ($response['total'] == 0) {
                    break;
                }

                foreach ($response['memberships'] as $membership) {
                    $user = $cacheUsers[$membership['userId']] ?? null;
                    if ($user === null) {
                        throw new \Exception('User not found', Exception::CODE_NOT_FOUND);
                    }

                    $memberships[] = new Membership(
                        $membership['$id'],
                        $team,
                        $user,
                        $membership['roles'],
                        $membership['confirm']
                    );

                    $lastDocument = $membership['$id'];
                }

                $this->callback($memberships);

                if (count($memberships) < $batchSize) {
                    break;
                }
            }
        }
    }

    protected function exportGroupDatabases(int $batchSize, array $resources): void
    {
        $handleExportEntityScopedResources = function (string $resourceKey, callable $callback) use ($resources) {
            foreach (Resource::ENTITY_TYPE_RESOURCE_MAP as $entityKey => $entityResource) {
                try {
                    if (\in_array($entityResource[$resourceKey], $resources)) {
                        $callback($entityKey, $entityResource);
                    }
                } catch (\Throwable $e) {
                    $this->addError(
                        new Exception(
                            $entityResource[$resourceKey],
                            Transfer::GROUP_DATABASES,
                            message: $e->getMessage(),
                            code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                            previous: $e
                        )
                    );

                    return false;
                }
            }
            return true;
        };

        try {
            if (Resource::isSupported(array_keys(Resource::DATABASE_TYPE_RESOURCE_MAP), $resources)) {
                $this->exportDatabases($batchSize, $resources);
            }
        } catch (\Throwable $e) {
            $this->addError(
                new Exception(
                    Resource::TYPE_DATABASE,
                    Transfer::GROUP_DATABASES,
                    message: $e->getMessage(),
                    code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                    previous: $e
                )
            );

            return;
        }

        foreach (Resource::DATABASE_TYPE_RESOURCE_MAP as $databaseKey => $databaseResource) {
            try {
                if (\in_array($databaseResource['entity'], $resources)) {
                    $this->exportEntities($databaseKey, $batchSize);
                }
            } catch (\Throwable $e) {
                $this->addError(
                    new Exception(
                        $databaseResource['entity'],
                        Transfer::GROUP_DATABASES,
                        message: $e->getMessage(),
                        code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                        previous: $e
                    )
                );

                return;
            }
        }

        // field
        if (!$handleExportEntityScopedResources('field', fn ($entityKey) => $this->exportFields($entityKey, $batchSize))) {
            return;
        }

        // index
        if (!$handleExportEntityScopedResources('index', fn ($entityKey) => $this->exportIndexes($entityKey, $batchSize))) {
            return;
        }

        // record
        if (!$handleExportEntityScopedResources('record', fn ($entityKey, $entityResource) => $this->exportRecords($entityKey, $entityResource['field'], $batchSize))) {
            return;
        }
    }

    /**
     * @param int $batchSize
     * @param array $resources
     * @throws Exception
     */
    private function exportDatabases(int $batchSize, array $resources = []): void
    {
        $lastDatabase = null;

        while (true) {
            $queries = [$this->reader->queryLimit($batchSize)];

            if ($this->rootResourceId !== '' && ($this->rootResourceType === Resource::TYPE_DATABASE || $this->rootResourceType === Resource::TYPE_DATABASE_DOCUMENTSDB)) {
                $targetDatabaseId = $this->rootResourceId;

                // Handle database:collection format - extract database ID
                if (\str_contains($this->rootResourceId, ':')) {
                    $parts = \explode(':', $this->rootResourceId, 2);
                    if (\count($parts) === 2) {
                        $targetDatabaseId = $parts[0];
                    }
                }

                $queries[] = $this->reader->queryEqual('$id', [$targetDatabaseId]);
                $queries[] = $this->reader->queryLimit(1);
            }

            $databases = [];

            if ($lastDatabase) {
                $queries[] = $this->reader->queryCursorAfter($lastDatabase);
            }

            $response = $this->reader->listDatabases($queries);

            foreach ($response as $database) {
                $originalType = $database['type'] ?? Resource::TYPE_DATABASE_LEGACY;

                $databaseFamilyType = $originalType;
                if (\in_array($databaseFamilyType, [Resource::TYPE_DATABASE_LEGACY, Resource::TYPE_DATABASE_TABLESDB], true)) {
                    $databaseFamilyType = Resource::TYPE_DATABASE;
                }

                if (Resource::isSupported($databaseFamilyType, $resources)) {
                    $newDatabase = self::getDatabase($databaseFamilyType, [
                        'id' => $database['$id'],
                        'name' => $database['name'],
                        'createdAt' => $database['$createdAt'],
                        'updatedAt' => $database['$updatedAt'],
                        'type' => $originalType,
                        'database' => $database['database'],
                        'enabled' => $database['enabled'] ?? true,
                    ]);
                    $databases[] = $newDatabase;

                }
            }

            if (empty($databases)) {
                break;
            }

            $lastDatabase = $databases[\count($databases) - 1];

            $this->callback($databases);

            if (\count($databases) < $batchSize) {
                break;
            }
        }
    }

    /**
     * @param string $databaseName
     * @param int $batchSize
     * @throws Exception
     */
    private function exportEntities(string $databaseName, int $batchSize): void
    {
        $databases = $this->cache->get($databaseName);
        foreach ($databases as $database) {
            /** @var Database $database */
            $lastTable = null;

            while (true) {
                $queries = [$this->reader->queryLimit($batchSize)];
                $tables = [];

                // Filter to specific table if rootResourceType is database with database:collection format
                if (
                    $this->rootResourceId !== '' &&
                    $this->rootResourceType === Resource::TYPE_DATABASE &&
                    \str_contains($this->rootResourceId, ':')
                ) {
                    $parts = \explode(':', $this->rootResourceId, 2);
                    if (\count($parts) === 2) {
                        $targetTableId = $parts[1]; // table ID
                        $queries[] = $this->reader->queryEqual('$id', [$targetTableId]);
                        $queries[] = $this->reader->queryLimit(1);
                    }
                } elseif (
                    $this->rootResourceId !== '' &&
                    $this->rootResourceType === Resource::TYPE_TABLE
                ) {
                    $targetTableId = $this->rootResourceId;
                    $queries[] = $this->reader->queryEqual('$id', [$targetTableId]);
                    $queries[] = $this->reader->queryLimit(1);
                } elseif ($lastTable) {
                    $queries[] = $this->reader->queryCursorAfter($lastTable);
                }

                $response = $this->reader->listTables($database, $queries);
                foreach ($response as $table) {
                    $newTable = self::getEntity($databaseName, [
                        'id' => $table['$id'],
                        'name' => $table['name'],
                        'documentSecurity' => $table['documentSecurity'],
                        'permissions' => $table['$permissions'],
                        'createdAt' => $table['$createdAt'],
                        'updatedAt' => $table['$updatedAt'],
                        'enabled' => $table['enabled'] ?? true,
                        'database' => [
                            'id' => $database->getId(),
                            'name' => $database->getDatabaseName(),
                            'type' => $database->getType(),
                            'database' => $database->getDatabase(),
                        ]
                    ]);

                    $tables[] = $newTable;
                }

                if (empty($tables)) {
                    break;
                }

                $this->callback($tables);

                $lastTable = $tables[\count($tables) - 1];

                if (\count($tables) < $batchSize) {
                    break;
                }
            }
        }
    }

    /**
     * @param string $entityType
     * @param int $batchSize
     * @throws Exception
     */
    private function exportFields(string $entityType, int $batchSize): void
    {
        $entities = $this->cache->get($entityType);
        // Transfer Indexes

        /** @var array<Table|Collection> $table */
        foreach ($entities as $table) {
            $lastColumn = null;

            while (true) {
                $queries = [$this->reader->queryLimit($batchSize)];
                $columns = [];

                if ($lastColumn) {
                    $queries[] = $this->reader->queryCursorAfter($lastColumn);
                }

                $response = $this->reader->listColumns($table, $queries);

                foreach ($response as $column) {
                    if (
                        $column['type'] === UtopiaDatabase::VAR_RELATIONSHIP
                        && $column['side'] === UtopiaDatabase::RELATION_SIDE_CHILD
                    ) {
                        continue;
                    }

                    /** @var Table $table */
                    $col = match($table->getDatabase()->getType()) {
                        Resource::TYPE_DATABASE_VECTORSDB => self::getColumn($table, $column)->getAttribute(),
                        default => self::getColumn($table, $column),
                    };

                    $columns[] = $col;
                }

                if (empty($columns)) {
                    break;
                }

                $this->callback($columns);

                $lastColumn = $columns[count($columns) - 1];

                if (count($columns) < $batchSize) {
                    break;
                }
            }
        }
    }

    /**
     * @param string $entityType
     * @param int $batchSize
     * @throws Exception
     */
    private function exportIndexes(string $entityType, int $batchSize): void
    {
        $entities = $this->cache->get($entityType);
        // Transfer Indexes
        foreach ($entities as $table) {
            /** @var Table $table */
            $lastIndex = null;

            while (true) {
                $queries = [$this->reader->queryLimit($batchSize)];
                $indexes = [];

                if ($lastIndex) {
                    $queries[] = $this->reader->queryCursorAfter($lastIndex);
                }

                $response = $this->reader->listIndexes($table, $queries);

                foreach ($response as $index) {
                    $indexes[] = new Index(
                        $index['$id'],
                        $index['key'],
                        $table,
                        $index['type'],
                        $index['columns'] ?? $index['attributes'],
                        $index['lengths'] ?? [],
                        $index['orders'] ?? [],
                        $index['$createdAt'] = empty($index['$createdAt']) ? UtopiaDateTime::now() : $index['$createdAt'],
                        $index['$updatedAt'] = empty($index['$updatedAt']) ? UtopiaDateTime::now() : $index['$updatedAt'],
                    );
                }

                if (empty($indexes)) {
                    break;
                }

                $this->callback($indexes);

                $lastIndex = $indexes[count($indexes) - 1];

                if (count($indexes) < $batchSize) {
                    break;
                }
            }
        }
    }

    /**
     * @param string $entityName
     * @param string $fieldName
     * @param int $batchSize
     * @throws Exception
     */
    private function exportRecords(string $entityName, string $fieldName, int $batchSize): void
    {
        $entities = $this->cache->get($entityName);

        foreach ($entities as $table) {
            /** @var Table $table */
            $lastRow = null;
            $iterationCount = 0;

            while (true) {
                $iterationCount++;

                $queries = [
                    $this->reader->queryLimit($batchSize),
                    ...$this->queries,
                ];

                $rows = [];

                if ($lastRow) {
                    $queries[] = $this->reader->queryCursorAfter($lastRow);
                }

                $selects = ['*', '$id', '$permissions', '$updatedAt', '$createdAt']; // We want relations flat!

                foreach ($selects as $select) {
                    $queries[] = $this->reader->querySelect($select);
                }

                $manyToMany = [];

                if ($this->reader->getSupportForAttributes()) {
                    $attributes = $this->cache->get($fieldName);

                    foreach ($attributes as $attribute) {
                        /** @var Relationship $attribute */
                        if (
                            $attribute->getTable()->getId() === $table->getId() &&
                            $attribute->getType() === Column::TYPE_RELATIONSHIP &&
                            $attribute->getSide() === 'parent' &&
                            $attribute->getRelationType() == 'manyToMany'
                        ) {
                            $manyToMany[] = $attribute->getKey();
                        }
                    }
                }

                $response = $this->reader->listRows($table, $queries);

                foreach ($response as $row) {
                    // HACK: Handle many to many (only for schema-based databases)
                    if (!empty($manyToMany)) {
                        $queries = [];
                        $queries[] = $this->reader->querySelect('$id'); // Adding $id because we can't select only relations

                        foreach ($manyToMany as $relation) {
                            $queries[] = $this->reader->querySelect($relation . '.$id');
                        }

                        $rowItem = $this->reader->getRow(
                            $table,
                            $row['$id'],
                            $queries
                        );

                        foreach ($manyToMany as $key) {
                            $row[$key] = [];
                            if (isset($rowItem[$key]) && is_array($rowItem[$key])) {
                                foreach ($rowItem[$key] as $relatedRowItem) {
                                    if (is_array($relatedRowItem) && isset($relatedRowItem['$id'])) {
                                        $row[$key][] = $relatedRowItem['$id'];
                                    }
                                }
                            }
                        }
                    }

                    $id = $row['$id'];
                    $permissions = $row['$permissions'];

                    unset($row['$id']);
                    unset($row['$permissions']);
                    unset($row['$collectionId']);
                    unset($row['$databaseId']);
                    unset($row['$sequence']);
                    unset($row['$collection']);

                    $row = self::getRecord($table->getDatabase()->getType(), [
                        'id' => $id,
                        'table' => [
                            'id' => $table->getId(),
                            'name' => $table->getTableName(),
                            'rowSecurity' => $table->getRowSecurity(),
                            'permissions' => $table->getPermissions(),
                            'database' => [
                                'id' => $table->getDatabase()->getId(),
                                'name' => $table->getDatabase()->getDatabaseName(),
                                'type' => $table->getDatabase()->getType(),
                                'database' => $table->getDatabase()->getDatabase(),
                            ]
                        ],
                        'data' => $row,
                        'permissions' => $permissions
                    ]);

                    $rows[] = $row;
                    $lastRow = $row;
                }

                $this->callback($rows);

                if (count($response) < $batchSize) {
                    break;
                }
            }
        }
    }

    protected function exportGroupStorage(int $batchSize, array $resources): void
    {
        try {
            if (\in_array(Resource::TYPE_BUCKET, $resources)) {
                $this->exportBuckets($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(
                new Exception(
                    Resource::TYPE_BUCKET,
                    Transfer::GROUP_STORAGE,
                    message: $e->getMessage(),
                    code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                    previous: $e
                )
            );
        }

        try {
            if (\in_array(Resource::TYPE_FILE, $resources)) {
                $this->exportFiles($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(
                new Exception(
                    Resource::TYPE_FILE,
                    Transfer::GROUP_STORAGE,
                    message: $e->getMessage(),
                    code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                    previous: $e
                )
            );
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportBuckets(int $batchSize): void
    {
        $queries = [];

        if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_BUCKET) {
            $queries[] = Query::equal('$id', $this->rootResourceId);
            $queries[] = Query::limit(1);
        }

        $buckets = $this->storage->listBuckets($queries);

        $convertedBuckets = [];

        foreach ($buckets['buckets'] as $bucket) {
            $bucket = new Bucket(
                $bucket['$id'],
                $bucket['name'],
                $bucket['$permissions'],
                $bucket['fileSecurity'],
                $bucket['enabled'],
                $bucket['maximumFileSize'],
                $bucket['allowedFileExtensions'],
                $bucket['compression'],
                $bucket['encryption'],
                $bucket['antivirus'],
                false,
                $bucket['transformations'] ?? false,
            );
            $convertedBuckets[] = $bucket;
        }

        if (empty($convertedBuckets)) {
            return;
        }

        $this->callback($convertedBuckets);
    }

    /**
     * @throws AppwriteException
     */
    private function exportFiles(int $batchSize): void
    {
        $buckets = $this->cache->get(Bucket::getName());

        foreach ($buckets as $bucket) {
            /** @var Bucket $bucket */
            $lastDocument = null;

            while (true) {
                $queries = [Query::limit($batchSize)];

                if ($lastDocument) {
                    $queries[] = Query::cursorAfter($lastDocument);
                }

                $response = $this->storage->listFiles(
                    $bucket->getId(),
                    $queries
                );

                foreach ($response['files'] as $file) {
                    try {
                        $this->exportFileData(new File(
                            $file['$id'],
                            $bucket,
                            $file['name'],
                            $file['signature'],
                            $file['mimeType'],
                            $file['$permissions'],
                            $file['sizeOriginal'],
                        ));
                    } catch (\Throwable $e) {
                        $this->addError(new Exception(
                            resourceName: Resource::TYPE_FILE,
                            resourceGroup: Transfer::GROUP_STORAGE,
                            resourceId: $file['$id'],
                            message: $e->getMessage(),
                            code: $e->getCode()
                        ));
                    }

                    $lastDocument = $file['$id'];
                }

                if (count($response['files']) < $batchSize) {
                    break;
                }
            }
        }
    }

    /**
     * @throws \Exception
     */
    private function exportFileData(File $file): void
    {
        // Set the chunk size (5MB)
        $start = 0;
        $end = Transfer::STORAGE_MAX_CHUNK_SIZE - 1;

        // Get the file size
        $fileSize = $file->getSize();

        if ($end >= $fileSize) {
            $end = $fileSize - 1;
        }

        // Loop until the entire file is downloaded
        while ($start < $fileSize) {
            $chunkData = $this->call(
                'GET',
                "/storage/buckets/{$file->getBucket()->getId()}/files/{$file->getId()}/download",
                ['range' => "bytes=$start-$end"]
            );

            // Send the chunk to the callback function
            $file
                ->setData($chunkData)
                ->setStart($start)
                ->setEnd($end);

            $this->callback([$file]);

            // Update the range
            $start += Transfer::STORAGE_MAX_CHUNK_SIZE;
            $end += Transfer::STORAGE_MAX_CHUNK_SIZE;

            if ($end >= $fileSize) {
                $end = $fileSize - 1;
            }
        }
    }

    protected function exportGroupFunctions(int $batchSize, array $resources): void
    {
        try {
            if (\in_array(Resource::TYPE_FUNCTION, $resources)) {
                $this->exportFunctions($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_FUNCTION,
                Transfer::GROUP_FUNCTIONS,
                message: $e->getMessage(),
                code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                previous: $e
            ));
        }

        try {
            $exportOnlyActive = !\in_array(Resource::TYPE_DEPLOYMENT, $resources);
            $this->exportDeployments($batchSize, $exportOnlyActive);
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_DEPLOYMENT,
                Transfer::GROUP_FUNCTIONS,
                message: $e->getMessage(),
                code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                previous: $e
            ));
        }
    }

    protected function exportGroupSites(int $batchSize, array $resources): void
    {
        try {
            if (\in_array(Resource::TYPE_SITE, $resources)) {
                $this->exportSites($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_SITE,
                Transfer::GROUP_SITES,
                message: $e->getMessage(),
                code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                previous: $e
            ));
        }

        try {
            $exportOnlyActive = !\in_array(Resource::TYPE_SITE_DEPLOYMENT, $resources);
            $this->exportSiteDeployments($batchSize, $exportOnlyActive);
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_SITE_DEPLOYMENT,
                Transfer::GROUP_SITES,
                message: $e->getMessage(),
                code: (int) $e->getCode() ?: Exception::CODE_INTERNAL,
                previous: $e
            ));
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportFunctions(int $batchSize): void
    {
        $this->functions = new Functions($this->client);
        /**
         * @var Func|null $lastFunction
         */
        $lastFunction = null;
        while (true) {
            $queries = [Query::limit($batchSize)];

            if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_FUNCTION) {
                $queries[] = Query::equal('$id', $this->rootResourceId);
                $queries[] = Query::limit(1);
            }

            if ($lastFunction) {
                $queries[] = Query::cursorAfter($lastFunction->getId());
            }

            $response = $this->functions->list($queries);


            if ($response['total'] === 0) {
                return;
            }

            $functions = [];
            $convertedResources = [];

            foreach ($response['functions'] as $function) {
                $convertedFunc = new Func(
                    $function['$id'],
                    $function['name'],
                    $function['runtime'],
                    $function['execute'],
                    $function['enabled'],
                    $function['events'],
                    $function['schedule'],
                    $function['timeout'],
                    $function['deploymentId'] ?? '',
                    $function['entrypoint'],
                    $function['commands'] ?? '',
                    $function['logging'] ?? true,
                    $function['scopes'] ?? [],
                    $function['specification'] ?? '',
                );
                $functions[] = $convertedFunc;

                $convertedResources[] = $convertedFunc;

                foreach ($function['vars'] as $var) {
                    $convertedResources[] = new EnvVar(
                        $var['$id'],
                        $convertedFunc,
                        $var['key'],
                        $var['value'],
                    );
                }
            }

            $lastFunction = $functions[count($functions) - 1];

            $this->callback($convertedResources);
            if (count($functions) < $batchSize) {
                return;
            }
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportDeployments(int $batchSize, bool $exportOnlyActive = false): void
    {
        $this->functions = new Functions($this->client);
        $functions = $this->cache->get(Func::getName());

        foreach ($functions as $func) {
            /** @var Func $func */
            $lastDocument = null;

            if ($exportOnlyActive) {
                $activeDeploymentId = $func->getActiveDeployment();
                if (empty($activeDeploymentId)) {
                    continue; // active-only mode: nothing to export for this function
                }

                $deployment = $this->functions->getDeployment($func->getId(), $activeDeploymentId);

                try {
                    $this->exportDeploymentData($func, $deployment);
                } catch (\Throwable $e) {
                    $func->setStatus(Resource::STATUS_ERROR, $e->getMessage());
                }

                continue;
            }

            while (true) {
                $queries = [Query::limit($batchSize)];

                if ($lastDocument) {
                    $queries[] = Query::cursorAfter($lastDocument);
                }

                $response = $this->functions->listDeployments(
                    $func->getId(),
                    $queries
                );

                foreach ($response['deployments'] as $deployment) {
                    try {
                        $this->exportDeploymentData($func, $deployment);
                    } catch (\Throwable $e) {
                        $func->setStatus(Resource::STATUS_ERROR, $e->getMessage());
                    }

                    $lastDocument = $deployment['$id'];
                }

                if (count($response['deployments']) < $batchSize) {
                    break;
                }
            }
        }
    }

    /**
     * @throws \Exception
     */
    private function exportDeploymentData(Func $func, array $deployment): void
    {
        // Set the chunk size (5MB)
        $start = 0;
        $end = Transfer::STORAGE_MAX_CHUNK_SIZE - 1;

        // Get the file size
        $responseHeaders = [];

        $this->call(
            'HEAD',
            "/functions/{$func->getId()}/deployments/{$deployment['$id']}/download",
            [],
            [],
            $responseHeaders
        );

        // content-length header missing, file is less than max buffer size
        if (!array_key_exists('content-length', $responseHeaders)) {
            $file = $this->call(
                'GET',
                "/functions/{$func->getId()}/deployments/{$deployment['$id']}/download",
                [],
                [],
                $responseHeaders
            );

            $size = strlen($file);

            if ($end > $size) {
                $end = $size - 1;
            }

            $deployment = new Deployment(
                $deployment['$id'],
                $func,
                $size,
                $deployment['entrypoint'],
                $start,
                $end,
                $file,
                $deployment['activate']
            );
            $deployment->setSequence($deployment->getId());

            $this->callback([$deployment]);

            return;
        }

        $fileSize = $responseHeaders['content-length'];

        if ($end >= $fileSize) {
            $end = $fileSize - 1;
        }

        $deployment = new Deployment(
            $deployment['$id'],
            $func,
            $fileSize,
            $deployment['entrypoint'],
            $start,
            $end,
            '',
            $deployment['activate']
        );

        $deployment->setSequence($deployment->getId());

        // Loop until the entire file is downloaded
        while ($start < $fileSize) {
            $chunkData = $this->call(
                'GET',
                "/functions/{$func->getId()}/deployments/{$deployment->getSequence()}/download",
                ['range' => "bytes=$start-$end"]
            );

            // Send the chunk to the callback function
            $deployment
                ->setData($chunkData)
                ->setStart($start)
                ->setEnd($end);

            $this->callback([$deployment]);

            // Update the range
            $start += Transfer::STORAGE_MAX_CHUNK_SIZE;
            $end += Transfer::STORAGE_MAX_CHUNK_SIZE;

            if ($end >= $fileSize) {
                $end = $fileSize - 1;
            }
        }
    }

    /**
     * @param array<string> $resources
     * @param array<string, int> $report
     * @param array<string, array<string>> $resourceIds
     * @throws AppwriteException
     */
    private function reportMessaging(array $resources, array &$report, array $resourceIds = []): void
    {
        if (\in_array(Resource::TYPE_PROVIDER, $resources)) {
            $providerQueries = $this->buildQueries(
                resourceType: Resource::TYPE_PROVIDER,
                resourceIds: $resourceIds,
                limit: 1
            );
            $report[Resource::TYPE_PROVIDER] = $this->messaging->listProviders($providerQueries)['total'];
        }

        if (\in_array(Resource::TYPE_TOPIC, $resources)) {
            $topicQueries = $this->buildQueries(
                resourceType: Resource::TYPE_TOPIC,
                resourceIds: $resourceIds,
                limit: 1
            );
            $report[Resource::TYPE_TOPIC] = $this->messaging->listTopics($topicQueries)['total'];
        }

        if (\in_array(Resource::TYPE_SUBSCRIBER, $resources)) {
            $subscriberTotal = 0;
            $lastTopic = null;

            while (true) {
                $topicQueries = [Query::limit(self::DEFAULT_PAGE_LIMIT)];
                if ($lastTopic) {
                    $topicQueries[] = Query::cursorAfter($lastTopic);
                }

                $topicResponse = $this->messaging->listTopics($topicQueries);
                if ($topicResponse['total'] == 0 || empty($topicResponse['topics'])) {
                    break;
                }

                foreach ($topicResponse['topics'] as $topic) {
                    $subscriberTotal += $this->messaging->listSubscribers($topic['$id'], [Query::limit(1)])['total'];
                    $lastTopic = $topic['$id'];
                }

                if (\count($topicResponse['topics']) < self::DEFAULT_PAGE_LIMIT) {
                    break;
                }
            }

            $report[Resource::TYPE_SUBSCRIBER] = $subscriberTotal;
        }

        if (\in_array(Resource::TYPE_MESSAGE, $resources)) {
            $messageQueries = $this->buildQueries(
                resourceType: Resource::TYPE_MESSAGE,
                resourceIds: $resourceIds,
                limit: 1
            );
            $report[Resource::TYPE_MESSAGE] = $this->messaging->listMessages($messageQueries)['total'];
        }
    }

    protected function exportGroupMessaging(int $batchSize, array $resources): void
    {
        try {
            if (\in_array(Resource::TYPE_PROVIDER, $resources)) {
                $this->exportProviders($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_PROVIDER,
                Transfer::GROUP_MESSAGING,
                message: $e->getMessage(),
                code: $e->getCode(),
                previous: $e
            ));
        }

        try {
            if (\in_array(Resource::TYPE_TOPIC, $resources)) {
                $this->exportTopics($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_TOPIC,
                Transfer::GROUP_MESSAGING,
                message: $e->getMessage(),
                code: $e->getCode(),
                previous: $e
            ));
        }

        try {
            if (\in_array(Resource::TYPE_SUBSCRIBER, $resources)) {
                $this->exportSubscribers($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_SUBSCRIBER,
                Transfer::GROUP_MESSAGING,
                message: $e->getMessage(),
                code: $e->getCode(),
                previous: $e
            ));
        }

        try {
            if (\in_array(Resource::TYPE_MESSAGE, $resources)) {
                $this->exportMessages($batchSize);
            }
        } catch (\Throwable $e) {
            $this->addError(new Exception(
                Resource::TYPE_MESSAGE,
                Transfer::GROUP_MESSAGING,
                message: $e->getMessage(),
                code: $e->getCode(),
                previous: $e
            ));
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportProviders(int $batchSize): void
    {
        $lastDocument = null;

        while (true) {
            $providers = [];

            $queries = [Query::limit($batchSize)];

            if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_PROVIDER) {
                $queries[] = Query::equal('$id', $this->rootResourceId);
                $queries[] = Query::limit(1);
            }

            if ($lastDocument) {
                $queries[] = Query::cursorAfter($lastDocument);
            }

            $response = $this->messaging->listProviders($queries);

            if ($response['total'] == 0) {
                break;
            }

            foreach ($response['providers'] as $provider) {
                $providers[] = new Provider(
                    $provider['$id'],
                    $provider['name'],
                    $provider['provider'],
                    $provider['type'],
                    $provider['enabled'],
                    $provider['credentials'] ?? [],
                    $provider['options'] ?? [],
                    $provider['$createdAt'] ?? '',
                    $provider['$updatedAt'] ?? '',
                );

                $lastDocument = $provider['$id'];
            }

            $this->callback($providers);

            if (\count($providers) < $batchSize) {
                break;
            }
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportTopics(int $batchSize): void
    {
        $lastDocument = null;

        while (true) {
            $topics = [];

            $queries = [Query::limit($batchSize)];

            if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_TOPIC) {
                $queries[] = Query::equal('$id', $this->rootResourceId);
                $queries[] = Query::limit(1);
            }

            if ($lastDocument) {
                $queries[] = Query::cursorAfter($lastDocument);
            }

            $response = $this->messaging->listTopics($queries);

            if ($response['total'] == 0) {
                break;
            }

            foreach ($response['topics'] as $topic) {
                $topics[] = new Topic(
                    $topic['$id'],
                    $topic['name'],
                    $topic['subscribe'] ?? [],
                    $topic['$createdAt'] ?? '',
                    $topic['$updatedAt'] ?? '',
                );

                $lastDocument = $topic['$id'];
            }

            $this->callback($topics);

            if (\count($topics) < $batchSize) {
                break;
            }
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportSubscribers(int $batchSize): void
    {
        $topics = $this->cache->get(Topic::getName());

        foreach ($topics as $topic) {
            /** @var Topic $topic */
            $lastDocument = null;

            while (true) {
                $subscribers = [];

                $queries = [Query::limit($batchSize)];

                if ($lastDocument) {
                    $queries[] = Query::cursorAfter($lastDocument);
                }

                $response = $this->messaging->listSubscribers($topic->getId(), $queries);

                if ($response['total'] == 0) {
                    break;
                }

                foreach ($response['subscribers'] as $subscriber) {
                    $subscribers[] = new Subscriber(
                        $subscriber['$id'],
                        $subscriber['topicId'],
                        $subscriber['targetId'],
                        $subscriber['userId'] ?? '',
                        $subscriber['userName'] ?? '',
                        $subscriber['providerType'] ?? '',
                        $subscriber['$createdAt'] ?? '',
                        $subscriber['$updatedAt'] ?? '',
                    );

                    $lastDocument = $subscriber['$id'];
                }

                $this->callback($subscribers);

                if (\count($subscribers) < $batchSize) {
                    break;
                }
            }
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportMessages(int $batchSize): void
    {
        $lastDocument = null;

        while (true) {
            $messages = [];

            $queries = [Query::limit($batchSize)];

            if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_MESSAGE) {
                $queries[] = Query::equal('$id', $this->rootResourceId);
                $queries[] = Query::limit(1);
            }

            if ($lastDocument) {
                $queries[] = Query::cursorAfter($lastDocument);
            }

            $response = $this->messaging->listMessages($queries);

            if ($response['total'] == 0) {
                break;
            }

            foreach ($response['messages'] as $message) {
                $messages[] = new Message(
                    $message['$id'],
                    $message['providerType'] ?? '',
                    $message['topics'] ?? [],
                    $message['users'] ?? [],
                    $message['targets'] ?? [],
                    $message['data'] ?? [],
                    $message['status'] ?? '',
                    $message['scheduledAt'] ?? '',
                    $message['deliveredAt'] ?? '',
                    $message['deliveryErrors'] ?? [],
                    $message['deliveredTotal'] ?? 0,
                    $message['$createdAt'] ?? '',
                    $message['$updatedAt'] ?? '',
                );

                $lastDocument = $message['$id'];
            }

            $this->callback($messages);

            if (\count($messages) < $batchSize) {
                break;
            }
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportSites(int $batchSize): void
    {
        $this->sites = new Sites($this->client);
        /** @var Site|null $lastSite */
        $lastSite = null;

        while (true) {
            $queries = [Query::limit($batchSize)];

            if ($this->rootResourceId !== '' && $this->rootResourceType === Resource::TYPE_SITE) {
                $queries[] = Query::equal('$id', $this->rootResourceId);
                $queries[] = Query::limit(1);
            }

            if ($lastSite) {
                $queries[] = Query::cursorAfter($lastSite->getId());
            }

            $response = $this->sites->list($queries);

            if ($response['total'] === 0) {
                return;
            }

            $sites = [];
            $convertedResources = [];

            foreach ($response['sites'] as $site) {
                $convertedSite = new Site(
                    $site['$id'],
                    $site['name'],
                    $site['framework'],
                    $site['buildRuntime'],
                    $site['enabled'],
                    $site['logging'],
                    $site['timeout'],
                    $site['installCommand'] ?? '',
                    $site['buildCommand'] ?? '',
                    $site['outputDirectory'] ?? '',
                    $site['adapter'] ?? 'static',
                    $site['fallbackFile'] ?? '',
                    $site['specification'] ?? '',
                    $site['deploymentId'] ?? ''
                );
                $sites[] = $convertedSite;
                $convertedResources[] = $convertedSite;

                $variables = $this->sites->listVariables($site['$id']);
                foreach ($variables['variables'] ?? [] as $var) {
                    $convertedResources[] = new SiteEnvVar(
                        $var['$id'],
                        $convertedSite,
                        $var['key'],
                        $var['value']
                    );
                }
            }

            $lastSite = $sites[count($sites) - 1];

            $this->callback($convertedResources);

            if (count($sites) < $batchSize) {
                return;
            }
        }
    }

    /**
     * @throws AppwriteException
     */
    private function exportSiteDeployments(int $batchSize, bool $exportOnlyActive = false): void
    {
        $this->sites = new Sites($this->client);
        $sites = $this->cache->get(Site::getName());

        foreach ($sites as $site) {
            /** @var Site $site */
            $lastDocument = null;

            if ($exportOnlyActive) {
                $activeDeploymentId = $site->getActiveDeployment();
                if (empty($activeDeploymentId)) {
                    continue; // active-only mode: nothing to export for this site
                }

                $deployment = $this->sites->getDeployment($site->getId(), $activeDeploymentId);

                try {
                    $this->exportSiteDeploymentData($site, $deployment);
                } catch (\Throwable $e) {
                    $site->setStatus(Resource::STATUS_ERROR, $e->getMessage());
                }

                continue;
            }

            while (true) {
                $queries = [Query::limit($batchSize)];

                if ($lastDocument) {
                    $queries[] = Query::cursorAfter($lastDocument);
                }

                $response = $this->sites->listDeployments(
                    $site->getId(),
                    $queries
                );

                foreach ($response['deployments'] as $deployment) {
                    try {
                        $this->exportSiteDeploymentData($site, $deployment);
                    } catch (\Throwable $e) {
                        $site->setStatus(Resource::STATUS_ERROR, $e->getMessage());
                    }

                    $lastDocument = $deployment['$id'];
                }

                if (count($response['deployments']) < $batchSize) {
                    break;
                }
            }
        }
    }

    /**
     * @throws \Exception
     */
    private function exportSiteDeploymentData(Site $site, array $deployment): void
    {
        $start = 0;
        $end = Transfer::STORAGE_MAX_CHUNK_SIZE - 1;

        $responseHeaders = [];

        $this->call(
            'HEAD',
            "/sites/{$site->getId()}/deployments/{$deployment['$id']}/download",
            [],
            [],
            $responseHeaders
        );

        if (!\array_key_exists('content-length', $responseHeaders)) {
            $file = $this->call(
                'GET',
                "/sites/{$site->getId()}/deployments/{$deployment['$id']}/download",
                [],
                [],
                $responseHeaders
            );

            $size = strlen($file);

            if ($end > $size) {
                $end = $size - 1;
            }

            $siteDeployment = new SiteDeployment(
                $deployment['$id'],
                $site,
                $size,
                $start,
                $end,
                $file,
                $deployment['$id'] === $site->getActiveDeployment()
            );
            $siteDeployment->setSequence($siteDeployment->getId());

            $this->callback([$siteDeployment]);

            return;
        }

        $fileSize = $responseHeaders['content-length'];

        if ($end >= $fileSize) {
            $end = $fileSize - 1;
        }

        $siteDeployment = new SiteDeployment(
            $deployment['$id'],
            $site,
            $fileSize,
            $start,
            $end,
            '',
            $deployment['$id'] === $site->getActiveDeployment()
        );

        $siteDeployment->setSequence($siteDeployment->getId());

        while ($start < $fileSize) {
            $chunkData = $this->call(
                'GET',
                "/sites/{$site->getId()}/deployments/{$siteDeployment->getSequence()}/download",
                ['range' => "bytes=$start-$end"]
            );

            $siteDeployment
                ->setData($chunkData)
                ->setStart($start)
                ->setEnd($end);

            $this->callback([$siteDeployment]);

            $start += Transfer::STORAGE_MAX_CHUNK_SIZE;
            $end += Transfer::STORAGE_MAX_CHUNK_SIZE;

            if ($end >= $fileSize) {
                $end = $fileSize - 1;
            }
        }
    }

    /**
     * @param string $databaseType
     * @param array $database {
     *     id: string,
     *     name: string,
     *     createdAt: string,
     *     updatedAt: string,
     *     enabled: bool,
     *     originalId: string|null,
     *     database: string
     * }
    */
    public static function getDatabase(string $databaseType, array $database): Resource
    {
        switch ($databaseType) {
            case Resource::TYPE_DATABASE_DOCUMENTSDB:
                return DocumentsDB::fromArray($database);
            case Resource::TYPE_DATABASE_VECTORSDB:
                return VectorsDB::fromArray($database);
            default:
                return Database::fromArray($database);
        }
    }

    /**
     * eg., tables,collections
     * @param string $databaseType
     * @param array{
     *     database: array{
     *        id: string,
     *        name: string,
     *     },
     *     name: string,
     *     id: string,
     *     documentSecurity?: bool,
     *     rowSecurity?: bool,
     *     permissions: ?array<string>,
     *     createdAt: string,
     *     updatedAt: string,
     *     enabled: bool
     * } $entity
     */
    public static function getEntity(string $databaseType, array $entity): Resource
    {
        switch ($databaseType) {
            case Resource::TYPE_DATABASE_DOCUMENTSDB:
                return Collection::fromArray($entity);
            case Resource::TYPE_DATABASE_VECTORSDB:
                return Collection::fromArray($entity);
            default:
                return Table::fromArray($entity);
        }
    }

    /**
     * eg.,documents/attributes
     * @param string $databaseType
     * @param array{
     *     id: string,
     *     collection?: array{
     *         database: array{
     *             id: string,
     *             name: string,
     *         },
     *         name: string,
     *         id: string,
     *         documentSecurity: bool,
     *         permissions: ?array<string>
     *     },
     *     table?: array{
     *         database: array{
     *             id: string,
     *             name: string,
     *         },
     *         name: string,
     *         id: string,
     *         rowSecurity: bool,
     *         permissions: ?array<string>
     *     },
     *     data: array<string, mixed>,
     *     permissions: ?array<string>
     * } $record
     */
    public static function getRecord(string $databaseType, array $record): Resource
    {
        switch ($databaseType) {
            case Resource::TYPE_DATABASE_DOCUMENTSDB:
                return Document::fromArray($record);
            case Resource::TYPE_DATABASE_VECTORSDB:
                return Document::fromArray($record);
            default:
                return Row::fromArray($record);
        }
    }

    public static function getColumn(Table $table, mixed $column): Column
    {
        return match ($column['type']) {
            Column::TYPE_STRING => match ($column['format'] ?? '') {
                Column::TYPE_EMAIL => new Email(
                    $column['key'],
                    $table,
                    required: $column['required'],
                    default: $column['default'],
                    array: $column['array'],
                    size: $column['size'] ?? 254,
                    createdAt: $column['$createdAt'] ?? '',
                    updatedAt: $column['$updatedAt'] ?? '',
                ),
                Column::TYPE_ENUM => new Enum(
                    $column['key'],
                    $table,
                    elements: $column['elements'],
                    required: $column['required'],
                    default: $column['default'],
                    array: $column['array'],
                    size: $column['size'] ?? UtopiaDatabase::LENGTH_KEY,
                    createdAt: $column['$createdAt'] ?? '',
                    updatedAt: $column['$updatedAt'] ?? '',
                ),
                Column::TYPE_URL => new URL(
                    $column['key'],
                    $table,
                    required: $column['required'],
                    default: $column['default'],
                    array: $column['array'],
                    size: $column['size'] ?? 2000,
                    createdAt: $column['$createdAt'] ?? '',
                    updatedAt: $column['$updatedAt'] ?? '',
                ),
                Column::TYPE_IP => new IP(
                    $column['key'],
                    $table,
                    required: $column['required'],
                    default: $column['default'],
                    array: $column['array'],
                    size: $column['size'] ?? 39,
                    createdAt: $column['$createdAt'] ?? '',
                    updatedAt: $column['$updatedAt'] ?? '',
                ),
                default => new Text(
                    $column['key'],
                    $table,
                    required: $column['required'],
                    default: $column['default'],
                    array: $column['array'],
                    size: $column['size'] ?? 0,
                    createdAt: $column['$createdAt'] ?? '',
                    updatedAt: $column['$updatedAt'] ?? '',
                ),
            },

            Column::TYPE_BOOLEAN => new Boolean(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_INTEGER => new Integer(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                min: $column['min'] ?? null,
                max: $column['max'] ?? null,
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_FLOAT => new Decimal(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                min: $column['min'] ?? null,
                max: $column['max'] ?? null,
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_RELATIONSHIP => new Relationship(
                $column['key'],
                $table,
                relatedTable: $column['relatedTable'] ?? $column['relatedCollection'],
                relationType: $column['relationType'],
                twoWay: $column['twoWay'],
                twoWayKey: $column['twoWayKey'],
                onDelete: $column['onDelete'],
                side: $column['side'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_DATETIME => new DateTime(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_POINT => new Point(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_LINE => new Line(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_POLYGON => new Polygon(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_OBJECT => new ObjectType(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_VECTOR => new Vector(
                $column['key'],
                $table,
                size: $column['size'],
                required: $column['required'],
                default: $column['default'],
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_VARCHAR => new Varchar(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                size: $column['size'] ?? 255,
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_TEXT => new RegularText(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                size: $column['size'] ?? 65535,
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_MEDIUMTEXT => new MediumText(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                size: $column['size'] ?? 16777215,
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),

            Column::TYPE_LONGTEXT => new LongText(
                $column['key'],
                $table,
                required: $column['required'],
                default: $column['default'],
                array: $column['array'],
                size: $column['size'] ?? 2147483647,
                createdAt: $column['$createdAt'] ?? '',
                updatedAt: $column['$updatedAt'] ?? '',
            ),


            default => throw new \InvalidArgumentException("Unsupported column type: {$column['type']}"),
        };

    }

    /**
     * Convert a raw attribute array to an Attribute resource.
     * This is now a thin wrapper that uses getColumn() and then converts to Attribute.
     * Since Collection extends Table, we can pass it directly to getColumn().
     *
     * @param Collection $collection
     * @param mixed $attribute
     * @return Attribute
     */
    public static function getAttribute(Collection $collection, mixed $attribute): Attribute
    {
        // Use getColumn() to create the Column resource, then convert to Attribute
        // This eliminates duplication - all type-specific logic is in getColumn()
        return self::getColumn($collection, $attribute)->getAttribute();
    }

    /**
     * Build queries with optional filtering by resource IDs
     */
    private function buildQueries(
        string $resourceType,
        array $resourceIds,
        ?string $cursor = null,
        int $limit = self::DEFAULT_PAGE_LIMIT
    ): array {
        $queries = [];

        if (!empty($resourceIds[$resourceType])) {
            $ids = (array) $resourceIds[$resourceType];

            $queries[] = Query::equal('$id', $ids);
        }

        if ($cursor) {
            $queries[] = Query::cursorAfter($cursor);
        } else {
            $queries[] = Query::limit($limit);
        }

        return $queries;
    }
}