diff --git a/app/realtime.php b/app/realtime.php index 86f9c85fdd..b2cef82522 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -15,10 +15,12 @@ use Swoole\Timer; use Utopia\Abuse\Abuse; use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis; use Utopia\App; +use Utopia\Cache\Adapter as CacheAdapter; use Utopia\Cache\Adapter\Sharding; use Utopia\Cache\Cache; use Utopia\CLI\Console; use Utopia\Config\Config; +use Utopia\Database\Adapter as DatabaseAdapter; use Utopia\Database\Database; use Utopia\Database\DateTime; use Utopia\Database\Document; @@ -41,100 +43,79 @@ require_once __DIR__ . '/init.php'; Runtime::enableCoroutine(SWOOLE_HOOK_ALL); // Allows overriding -if (!function_exists('getConsoleDB')) { - function getConsoleDB(): Database +if (!function_exists('useConsoleDB')) { + function useConsoleDB(): callable { global $register; /** @var \Utopia\Pools\Group $pools */ $pools = $register->get('pools'); - $dbAdapter = $pools - ->get('console') - ->pop() - ->getResource() - ; + return function (callable $callback) use ($pools): mixed { + return $pools->use( + ['console', ...Config::getParam('pools-cache')], + function (DatabaseAdapter $adapter, CacheAdapter ...$cacheAdapters) use ($callback) { + $database = new Database($adapter, new Cache(new Sharding($cacheAdapters))); + $database + ->setNamespace('_console') + ->setMetadata('host', \gethostname()) + ->setMetadata('project', '_console'); - $database = new Database($dbAdapter, getCache()); - - $database - ->setNamespace('_console') - ->setMetadata('host', \gethostname()) - ->setMetadata('project', '_console'); - - return $database; + return $callback($database); + } + ); + }; } } // Allows overriding -if (!function_exists('getProjectDB')) { - function getProjectDB(Document $project): Database +if (!function_exists('useProjectDB')) { + function useProjectDB(Document $project): callable { - global $register; - - /** @var \Utopia\Pools\Group $pools */ - $pools = $register->get('pools'); - if ($project->isEmpty() || $project->getId() === 'console') { - return getConsoleDB(); + return useConsoleDB(); } - try { - $dsn = new DSN($project->getAttribute('database')); - } catch (\InvalidArgumentException) { - // TODO: Temporary until all projects are using shared tables - $dsn = new DSN('mysql://' . $project->getAttribute('database')); - } - - $adapter = $pools - ->get($dsn->getHost()) - ->pop() - ->getResource(); - - $database = new Database($adapter, getCache()); - - $sharedTables = \explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', '')); - - if (\in_array($dsn->getHost(), $sharedTables)) { - $database - ->setSharedTables(true) - ->setTenant($project->getInternalId()) - ->setNamespace($dsn->getParam('namespace')); - } else { - $database - ->setSharedTables(false) - ->setTenant(null) - ->setNamespace('_' . $project->getInternalId()); - } - - $database - ->setMetadata('host', \gethostname()) - ->setMetadata('project', $project->getId()); - - return $database; - } -} - -// Allows overriding -if (!function_exists('getCache')) { - function getCache(): Cache - { global $register; - $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */ + /** @var \Utopia\Pools\Group $pools */ + $pools = $register->get('pools'); - $list = Config::getParam('pools-cache', []); - $adapters = []; + return function (callable $callback) use ($pools, $project) { + try { + $dsn = new DSN($project->getAttribute('database')); + } catch (\InvalidArgumentException) { + // TODO: Temporary until all projects are using shared tables + $dsn = new DSN('mysql://' . $project->getAttribute('database')); + } - foreach ($list as $value) { - $adapters[] = $pools - ->get($value) - ->pop() - ->getResource() - ; - } + return $pools->use( + [$dsn->getHost(), ...Config::getParam('pools-cache')], + function (DatabaseAdapter $adapter, CacheAdapter ...$cacheAdapters) use ($callback, $project, $dsn) { + $database = new Database($adapter, new Cache(new Sharding($cacheAdapters))); - return new Cache(new Sharding($adapters)); + $sharedTables = \explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', '')); + + if (\in_array($dsn->getHost(), $sharedTables)) { + $database + ->setSharedTables(true) + ->setTenant($project->getInternalId()) + ->setNamespace($dsn->getParam('namespace')); + } else { + $database + ->setSharedTables(false) + ->setTenant(null) + ->setNamespace('_' . $project->getInternalId()); + } + + $database + ->setMetadata('host', \gethostname()) + ->setMetadata('project', $project->getId()); + + return $callback($database); + } + ); + }; } } @@ -252,7 +233,6 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume */ go(function () use ($register, $containerId, &$statsDocument) { $attempts = 0; - $database = getConsoleDB(); do { try { @@ -266,7 +246,7 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume 'value' => '{}' ]); - $statsDocument = Authorization::skip(fn () => $database->createDocument('realtime', $document)); + $statsDocument = Authorization::skip(fn () => useConsoleDB()(fn (Database $database) => $database->createDocument('realtime', $document))); break; } catch (Throwable) { Console::warning("Collection not ready. Retrying connection ({$attempts})..."); @@ -291,13 +271,11 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume } try { - $database = getConsoleDB(); - $statsDocument ->setAttribute('timestamp', DateTime::now()) ->setAttribute('value', json_encode($payload)); - Authorization::skip(fn () => $database->updateDocument('realtime', $statsDocument->getId(), $statsDocument)); + Authorization::skip(fn () => useConsoleDB()(fn (Database $database) => $database->updateDocument('realtime', $statsDocument->getId(), $statsDocument))); } catch (Throwable $th) { call_user_func($logError, $th, "updateWorkerDocument"); } finally { @@ -326,13 +304,11 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, // TODO: Remove this if check once it doesn't cause issues for cloud if (System::getEnv('_APP_EDITION', 'self-hosted') === 'self-hosted') { if ($realtime->hasSubscriber('console', Role::users()->toString(), 'project')) { - $database = getConsoleDB(); - $payload = []; - $list = Authorization::skip(fn () => $database->find('realtime', [ + $list = Authorization::skip(fn () => useConsoleDB()(fn (Database $database) => $database->find('realtime', [ Query::greaterThan('timestamp', DateTime::addSeconds(new \DateTime(), -15)), - ])); + ]))); /** * Aggregate stats across containers. @@ -425,11 +401,9 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, if ($realtime->hasSubscriber($projectId, 'user:' . $userId)) { $connection = array_key_first(reset($realtime->subscriptions[$projectId]['user:' . $userId])); - $consoleDatabase = getConsoleDB(); - $project = Authorization::skip(fn () => $consoleDatabase->getDocument('projects', $projectId)); - $database = getProjectDB($project); + $project = Authorization::skip(fn () => useConsoleDB()(fn (Database $database) => $database->getDocument('projects', $projectId))); - $user = $database->getDocument('users', $userId); + $user = useProjectDB($project)(fn (Database $database) => $database->getDocument('users', $userId)); $roles = Auth::getRoles($user); $channels = $realtime->connections[$connection]['channels']; @@ -605,13 +579,11 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re try { $response = new Response(new SwooleResponse()); $projectId = $realtime->connections[$connection]['projectId']; - $database = getConsoleDB(); if ($projectId !== 'console') { - $project = Authorization::skip(fn () => $database->getDocument('projects', $projectId)); - $database = getProjectDB($project); + $project = Authorization::skip(fn () => useConsoleDB()(fn (Database $database) => $database->getDocument('projects', $projectId))); } else { - $project = null; + $project = new Document(); } /* @@ -653,7 +625,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re Auth::$unique = $session['id'] ?? ''; Auth::$secret = $session['secret'] ?? ''; - $user = $database->getDocument('users', Auth::$unique); + $user = useProjectDB($project)(fn (Database $database) => $database->getDocument('users', Auth::$unique)); if ( empty($user->getId()) // Check a document has been found in the DB diff --git a/composer.json b/composer.json index d7b8505b5c..6011ba7ffd 100644 --- a/composer.json +++ b/composer.json @@ -63,7 +63,7 @@ "utopia-php/migration": "0.8.*", "utopia-php/orchestration": "0.9.*", "utopia-php/platform": "0.7.*", - "utopia-php/pools": "0.7.*", + "utopia-php/pools": "0.8.*", "utopia-php/preloader": "0.2.*", "utopia-php/queue": "0.9.*", "utopia-php/registry": "0.5.*", diff --git a/composer.lock b/composer.lock index cdba756c72..94cbf01ed3 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "e0d7f21b681e4591144fec16c4f0d6aa", + "content-hash": "5e8f51c0f468cfde5384593116d2e541", "packages": [ { "name": "adhocore/jwt", @@ -4375,16 +4375,16 @@ }, { "name": "utopia-php/pools", - "version": "0.7.0", + "version": "0.8.0", "source": { "type": "git", "url": "https://github.com/utopia-php/pools.git", - "reference": "ad64d45afda08ec8b29e2642a8d18075964d40bf" + "reference": "60733929dc328e7ea47e800579c8bbf0d49df5ba" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/utopia-php/pools/zipball/ad64d45afda08ec8b29e2642a8d18075964d40bf", - "reference": "ad64d45afda08ec8b29e2642a8d18075964d40bf", + "url": "https://api.github.com/repos/utopia-php/pools/zipball/60733929dc328e7ea47e800579c8bbf0d49df5ba", + "reference": "60733929dc328e7ea47e800579c8bbf0d49df5ba", "shasum": "" }, "require": { @@ -4421,9 +4421,9 @@ ], "support": { "issues": "https://github.com/utopia-php/pools/issues", - "source": "https://github.com/utopia-php/pools/tree/0.7.0" + "source": "https://github.com/utopia-php/pools/tree/0.8.0" }, - "time": "2025-03-18T03:55:33+00:00" + "time": "2025-03-19T10:22:03+00:00" }, { "name": "utopia-php/preloader",