appwrite/app/test

$register->set('pools', function () {
    $group = new Group();

    $fallbackForDB = 'db_main=' . AppwriteURL::unparse([
        'scheme' => 'mariadb',
        'host' => App::getEnv('_APP_DB_HOST', 'mariadb'),
        'port' => App::getEnv('_APP_DB_PORT', '3306'),
        'user' => App::getEnv('_APP_DB_USER', ''),
        'pass' => App::getEnv('_APP_DB_PASS', ''),
        'path' => App::getEnv('_APP_DB_SCHEMA', ''),
    ]);
    $fallbackForRedis = 'redis_main=' . AppwriteURL::unparse([
        'scheme' => 'redis',
        'host' => App::getEnv('_APP_REDIS_HOST', 'redis'),
        'port' => App::getEnv('_APP_REDIS_PORT', '6379'),
        'user' => App::getEnv('_APP_REDIS_USER', ''),
        'pass' => App::getEnv('_APP_REDIS_PASS', ''),
    ]);

    $connections = [
        'console' => [
            'type' => 'database',
            'dsns' => App::getEnv('_APP_CONNECTIONS_DB_CONSOLE', $fallbackForDB),
            'multiple' => false,
            'schemes' => ['mariadb', 'mysql'],
        ],
        'database' => [
            'type' => 'database',
            'dsns' => App::getEnv('_APP_CONNECTIONS_DB_PROJECT', $fallbackForDB),
            'multiple' => true,
            'schemes' => ['mariadb', 'mysql'],
        ],
        'queue' => [
            'type' => 'queue',
            'dsns' => App::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis),
            'multiple' => false,
            'schemes' => ['redis'],
        ],
        'pubsub' => [
            'type' => 'pubsub',
            'dsns' => App::getEnv('_APP_CONNECTIONS_PUBSUB', $fallbackForRedis),
            'multiple' => false,
            'schemes' => ['redis'],
        ],
        'cache' => [
            'type' => 'cache',
            'dsns' => App::getEnv('_APP_CONNECTIONS_CACHE', $fallbackForRedis),
            'multiple' => true,
            'schemes' => ['redis'],
        ],
    ];

    $maxConnections = App::getEnv('_APP_CONNECTIONS_MAX', 151);
    $instanceConnections = $maxConnections / App::getEnv('_APP_POOL_CLIENTS', 14);

    $multiprocessing = App::getEnv('_APP_SERVER_MULTIPROCESS', 'disabled') === 'enabled';

    if ($multiprocessing) {
        $workerCount = swoole_cpu_num() * intval(App::getEnv('_APP_WORKER_PER_CORE', 6));
    } else {
        $workerCount = 1;
    }

    if ($workerCount > $instanceConnections) {
        throw new \Exception('Pool size is too small. Increase the number of allowed database connections or decrease the number of workers.', 500);
    }

    $poolSize = (int)($instanceConnections / $workerCount);

    foreach ($connections as $key => $connection) {
        $type = $connection['type'] ?? '';
        $dsns = $connection['dsns'] ?? '';
        $multipe = $connection['multiple'] ?? false;
        $schemes = $connection['schemes'] ?? [];
        $config = [];
        $dsns = explode(',', $connection['dsns'] ?? '');
        foreach ($dsns as &$dsn) {
            $dsn = explode('=', $dsn);
            $name = ($multipe) ? $key . '_' . $dsn[0] : $key;
            $dsn = $dsn[1] ?? '';
            $config[] = $name;
            if (empty($dsn)) {
                //throw new Exception(Exception::GENERAL_SERVER_ERROR, "Missing value for DSN connection in {$key}");
                continue;
            }

            $dsn = new DSN($dsn);
            $dsnHost = $dsn->getHost();
            $dsnPort = $dsn->getPort();
            $dsnUser = $dsn->getUser();
            $dsnPass = $dsn->getPassword();
            $dsnScheme = $dsn->getScheme();
            $dsnDatabase = $dsn->getPath();

            if (!in_array($dsnScheme, $schemes)) {
                throw new Exception(Exception::GENERAL_SERVER_ERROR, "Invalid console database scheme");
            }

            /**
             * Get Resource
             *
             * Creation could be reused accross connection types like database, cache, queue, etc.
             *
             * Resource assignment to an adapter will happen below.
             */
            switch ($dsnScheme) {
                case 'mysql':
                case 'mariadb':
                    $resource = function () use ($dsnHost, $dsnPort, $dsnUser, $dsnPass, $dsnDatabase) {
                        return new PDOProxy(function () use ($dsnHost, $dsnPort, $dsnUser, $dsnPass, $dsnDatabase) {
                            return new PDO("mysql:host={$dsnHost};port={$dsnPort};dbname={$dsnDatabase};charset=utf8mb4", $dsnUser, $dsnPass, array(
                                PDO::ATTR_TIMEOUT => 3, // Seconds
                                PDO::ATTR_PERSISTENT => true,
                                PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
                                PDO::ATTR_ERRMODE => App::isDevelopment() ? PDO::ERRMODE_WARNING : PDO::ERRMODE_SILENT, // If in production mode, warnings are not displayed
                                PDO::ATTR_EMULATE_PREPARES => true,
                                PDO::ATTR_STRINGIFY_FETCHES => true
                            ));
                        });
                    };
                    break;
                case 'redis':
                    $resource = function () use ($dsnHost, $dsnPort, $dsnPass) {
                        $redis = new Redis();
                        @$redis->pconnect($dsnHost, (int)$dsnPort);
                        if ($dsnPass) {
                            $redis->auth($dsnPass);
                        }
                        $redis->setOption(Redis::OPT_READ_TIMEOUT, -1);

                        return $redis;
                    };
                    break;

                default:
                    throw new Exception(Exception::GENERAL_SERVER_ERROR, "Invalid scheme");
                    break;
            }

            $pool = new Pool($name, $poolSize, function () use ($type, $resource, $dsn) {
                // Get Adapter
                $adapter = null;
                switch ($type) {
                    case 'database':
                        $adapter = match ($dsn->getScheme()) {
                            'mariadb' => new MariaDB($resource()),
                            'mysql' => new MySQL($resource()),
                            default => null
                        };

                        $adapter->setDefaultDatabase($dsn->getPath());
                        break;
                    case 'pubsub':
                        $adapter = $resource();
                        break;
                    case 'queue':
                        $adapter = match ($dsn->getScheme()) {
                            'redis' => new Queue\Connection\Redis($dsn->getHost(), $dsn->getPort()),
                            default => null
                        };
                        break;
                    case 'cache':
                        $adapter = match ($dsn->getScheme()) {
                            'redis' => new RedisCache($resource()),
                            default => null
                        };
                        break;

                    default:
                        throw new Exception(Exception::GENERAL_SERVER_ERROR, "Server error: Missing adapter implementation.");
                        break;
                }

                return $adapter;
            });

            $group->add($pool);
        }

        Config::setParam('pools-' . $key, $config);
    }

    return $group;
});