diff --git a/app/init/cli.php b/app/init/cli.php index c3602efd6a..cfa1fcbee2 100644 --- a/app/init/cli.php +++ b/app/init/cli.php @@ -15,10 +15,11 @@ require_once(__DIR__ . '/app.php'); */ require_once(__DIR__ . '/http.php'); - use Appwrite\Event\Certificate; use Appwrite\Event\Delete; use Appwrite\Event\Func; +use Appwrite\Event\StatsResources; +use Appwrite\Event\StatsUsage; use Appwrite\Runtimes\Runtimes; use Utopia\Cache\Adapter\Sharding; use Utopia\Cache\Cache; @@ -31,7 +32,7 @@ use Utopia\Database\Validator\Authorization; use Utopia\DSN\DSN; use Utopia\Logger\Log; use Utopia\Pools\Group; -use Utopia\Queue\Connection; +use Utopia\Queue\Publisher; use Utopia\Registry\Registry; use Utopia\System\System; @@ -39,7 +40,7 @@ use Utopia\System\System; Config::setParam('runtimes', (new Runtimes('v4'))->getAll(supported: false)); // require controllers after overwriting runtimes -require_once __DIR__ . '/../controllers/general.php'; +require_once __DIR__ . '/controllers/general.php'; Authorization::disable(); @@ -172,18 +173,57 @@ CLI::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform }; }, ['pools', 'dbForPlatform', 'cache']); -CLI::setResource('queue', function (Group $pools) { - return $pools->get('queue')->pop()->getResource(); +CLI::setResource('getLogsDB', function (Group $pools, Cache $cache) { + $database = null; + return function (?Document $project = null) use ($pools, $cache, $database) { + if ($database !== null && $project !== null && !$project->isEmpty() && $project->getId() !== 'console') { + $database->setTenant($project->getInternalId()); + return $database; + } + + $dbAdapter = $pools + ->get('logs') + ->pop() + ->getResource(); + + $database = new Database( + $dbAdapter, + $cache + ); + + $database + ->setSharedTables(true) + ->setNamespace('logsV1') + ->setTimeout(APP_DATABASE_TIMEOUT_MILLISECONDS) + ->setMaxQueryValues(APP_DATABASE_QUERY_MAX_VALUES); + + // set tenant + if ($project !== null && !$project->isEmpty() && $project->getId() !== 'console') { + $database->setTenant($project->getInternalId()); + } + + return $database; + }; +}, ['pools', 'cache']); + +CLI::setResource('queueForStatsUsage', function (Connection $publisher) { + return new StatsUsage($publisher); +}, ['publisher']); +CLI::setResource('queueForStatsResources', function (Publisher $publisher) { + return new StatsResources($publisher); +}, ['publisher']); +CLI::setResource('publisher', function (Group $pools) { + return $pools->get('publisher')->pop()->getResource(); }, ['pools']); -CLI::setResource('queueForFunctions', function (Connection $queue) { - return new Func($queue); -}, ['queue']); -CLI::setResource('queueForDeletes', function (Connection $queue) { - return new Delete($queue); -}, ['queue']); -CLI::setResource('queueForCertificates', function (Connection $queue) { - return new Certificate($queue); -}, ['queue']); +CLI::setResource('queueForFunctions', function (Publisher $publisher) { + return new Func($publisher); +}, ['publisher']); +CLI::setResource('queueForDeletes', function (Publisher $publisher) { + return new Delete($publisher); +}, ['publisher']); +CLI::setResource('queueForCertificates', function (Publisher $publisher) { + return new Certificate($publisher); +}, ['publisher']); CLI::setResource('logError', function (Registry $register) { return function (Throwable $error, string $namespace, string $action) use ($register) { $logger = $register->get('logger'); diff --git a/app/init/http.php b/app/init/http.php index 1c85c3d719..aa6b760bfd 100644 --- a/app/init/http.php +++ b/app/init/http.php @@ -43,7 +43,6 @@ use Utopia\DSN\DSN; use Utopia\Locale\Locale; use Utopia\Logger\Log; use Utopia\Pools\Group; -use Utopia\Queue\Connection; use Utopia\Storage\Device\Local; use Utopia\System\System; use Utopia\Validator\Hostname; diff --git a/app/init/worker.php b/app/init/worker.php index e9e0e3f9c3..41f5ed71bb 100644 --- a/app/init/worker.php +++ b/app/init/worker.php @@ -8,7 +8,6 @@ require_once(__DIR__ . '/app.php'); - use Appwrite\Certificates\LetsEncrypt; use Appwrite\Event\Audit; use Appwrite\Event\Build; @@ -20,8 +19,12 @@ use Appwrite\Event\Func; use Appwrite\Event\Mail; use Appwrite\Event\Messaging; use Appwrite\Event\Migration; +use Appwrite\Event\StatsUsage; +use Appwrite\Event\StatsUsageDump; +/** remove */ use Appwrite\Event\Usage; use Appwrite\Event\UsageDump; +/** /remove */ use Swoole\Runtime; use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis; use Utopia\Cache\Adapter\Sharding; @@ -35,8 +38,8 @@ use Utopia\Database\Validator\Authorization; use Utopia\DSN\DSN; use Utopia\Logger\Log; use Utopia\Pools\Group; -use Utopia\Queue\Connection; use Utopia\Queue\Message; +use Utopia\Queue\Publisher; use Utopia\Queue\Server; use Utopia\Registry\Registry; use Utopia\System\System; @@ -177,6 +180,39 @@ Server::setResource('getProjectDB', function (Group $pools, Database $dbForPlatf }; }, ['pools', 'dbForPlatform', 'cache']); +Server::setResource('getLogsDB', function (Group $pools, Cache $cache) { + $database = null; + return function (?Document $project = null) use ($pools, $cache, $database) { + if ($database !== null && $project !== null && !$project->isEmpty() && $project->getId() !== 'console') { + $database->setTenant($project->getInternalId()); + return $database; + } + + $dbAdapter = $pools + ->get('logs') + ->pop() + ->getResource(); + + $database = new Database( + $dbAdapter, + $cache + ); + + $database + ->setSharedTables(true) + ->setNamespace('logsV1') + ->setTimeout(APP_DATABASE_TIMEOUT_MILLISECONDS) + ->setMaxQueryValues(APP_DATABASE_QUERY_MAX_VALUES); + + // set tenant + if ($project !== null && !$project->isEmpty() && $project->getId() !== 'console') { + $database->setTenant($project->getInternalId()); + } + + return $database; + }; +}, ['pools', 'cache']); + Server::setResource('abuseRetention', function () { return time() - (int) System::getEnv('_APP_MAINTENANCE_RETENTION_ABUSE', 86400); }); @@ -228,57 +264,69 @@ Server::setResource('timelimit', function (\Redis $redis) { Server::setResource('log', fn () => new Log()); -Server::setResource('queueForUsage', function (Connection $queue) { - return new Usage($queue); -}, ['queue']); - -Server::setResource('queueForUsageDump', function (Connection $queue) { - return new UsageDump($queue); -}, ['queue']); - -Server::setResource('queue', function (Group $pools) { - return $pools->get('queue')->pop()->getResource(); +Server::setResource('publisher', function (Group $pools) { + return $pools->get('publisher')->pop()->getResource(); }, ['pools']); -Server::setResource('queueForDatabase', function (Connection $queue) { - return new EventDatabase($queue); -}, ['queue']); +Server::setResource('consumer', function (Group $pools) { + return $pools->get('consumer')->pop()->getResource(); +}, ['pools']); -Server::setResource('queueForMessaging', function (Connection $queue) { - return new Messaging($queue); -}, ['queue']); +Server::setResource('queueForUsage', function (Publisher $publisher) { + return new Usage($publisher); +}, ['publisher']); -Server::setResource('queueForMails', function (Connection $queue) { - return new Mail($queue); -}, ['queue']); +Server::setResource('queueForUsageDump', function (Publisher $publisher) { + return new UsageDump($publisher); +}, ['publisher']); -Server::setResource('queueForBuilds', function (Connection $queue) { - return new Build($queue); -}, ['queue']); +Server::setResource('queueForStatsUsage', function (Publisher $publisher) { + return new StatsUsage($publisher); +}, ['publisher']); -Server::setResource('queueForDeletes', function (Connection $queue) { - return new Delete($queue); -}, ['queue']); +Server::setResource('queueForStatsUsageDump', function (Publisher $publisher) { + return new StatsUsageDump($publisher); +}, ['publisher']); -Server::setResource('queueForEvents', function (Connection $queue) { - return new Event($queue); -}, ['queue']); +Server::setResource('queueForDatabase', function (Publisher $publisher) { + return new EventDatabase($publisher); +}, ['publisher']); -Server::setResource('queueForAudits', function (Connection $queue) { - return new Audit($queue); -}, ['queue']); +Server::setResource('queueForMessaging', function (Publisher $publisher) { + return new Messaging($publisher); +}, ['publisher']); -Server::setResource('queueForFunctions', function (Connection $queue) { - return new Func($queue); -}, ['queue']); +Server::setResource('queueForMails', function (Publisher $publisher) { + return new Mail($publisher); +}, ['publisher']); -Server::setResource('queueForCertificates', function (Connection $queue) { - return new Certificate($queue); -}, ['queue']); +Server::setResource('queueForBuilds', function (Publisher $publisher) { + return new Build($publisher); +}, ['publisher']); -Server::setResource('queueForMigrations', function (Connection $queue) { - return new Migration($queue); -}, ['queue']); +Server::setResource('queueForDeletes', function (Publisher $publisher) { + return new Delete($publisher); +}, ['publisher']); + +Server::setResource('queueForEvents', function (Publisher $publisher) { + return new Event($publisher); +}, ['publisher']); + +Server::setResource('queueForAudits', function (Publisher $publisher) { + return new Audit($publisher); +}, ['publisher']); + +Server::setResource('queueForFunctions', function (Publisher $publisher) { + return new Func($publisher); +}, ['publisher']); + +Server::setResource('queueForCertificates', function (Publisher $publisher) { + return new Certificate($publisher); +}, ['publisher']); + +Server::setResource('queueForMigrations', function (Publisher $publisher) { + return new Migration($publisher); +}, ['publisher']); Server::setResource('logger', function (Registry $register) { return $register->get('logger');