set('pools', function ($register) { return $register->get('pools'); }, ['register']); $container->set('authorization', function () { $authorization = new Authorization(); $authorization->disable(); return $authorization; }, []); $container->set('project', fn () => new Document([]), []); $container->set('consumer', function (Group $pools) { return new BrokerPool(consumer: $pools->get('consumer')); }, ['pools']); $container->set('consumerDatabases', function (BrokerPool $consumer) { return $consumer; }, ['consumer']); $container->set('consumerMigrations', function (BrokerPool $consumer) { return $consumer; }, ['consumer']); $container->set('consumerStatsUsage', function (BrokerPool $consumer) { return $consumer; }, ['consumer']); $container->set('certificates', function () { $email = System::getEnv('_APP_EMAIL_CERTIFICATES', System::getEnv('_APP_SYSTEM_SECURITY_EMAIL_ADDRESS')); if (empty($email)) { throw new Exception('You must set a valid security email address (_APP_EMAIL_CERTIFICATES) to issue a LetsEncrypt SSL certificate.'); } return new LetsEncrypt($email); }, []); $platform = new Appwrite(); $args = $_SERVER['argv'] ?? []; if (! isset($args[1])) { Console::error('Missing worker name'); Console::exit(1); } \array_shift($args); $workerName = $args[0]; if (\str_starts_with($workerName, 'databases')) { $queueName = System::getEnv('_APP_QUEUE_NAME', 'database_db_main'); } else { $queueName = System::getEnv('_APP_QUEUE_NAME', 'v1-' . strtolower($workerName)); } /** @var \Utopia\Pools\Group $pools */ $pools = $container->get('pools'); $adapter = new Swoole( $pools->get('consumer')->pop()->getResource(), System::getEnv('_APP_WORKERS_NUM', 1), $queueName ); $worker = new Server($adapter, $container); try { $worker->init()->action(function () use ($worker, $registerWorkerMessageResources) { $registerWorkerMessageResources($worker->getContainer()); }); $container->set('bus', function ($register) use ($worker) { return $register->get('bus')->setResolver( fn (string $name) => $worker->getContainer()->get($name) ); }, ['register']); $platform->setWorker($worker); $platform->init(Service::TYPE_WORKER, [ 'workerName' => strtolower($workerName), ]); } catch (\Throwable $e) { Console::error($e->getMessage() . ', File: ' . $e->getFile() . ', Line: ' . $e->getLine()); Console::exit(1); } $worker ->error() ->inject('error') ->inject('project') ->inject('authorization') ->action(function (Throwable $error, Document $project, Authorization $authorization) use ($queueName) { $version = System::getEnv('_APP_VERSION', 'UNKNOWN'); $action = 'appwrite-queue-' . $queueName; // Export through Span::init(), then restore the active worker span. $activeSpan = Span::current(); $span = Span::init($action); try { $span->set('level', 'error'); $span->set('logger', 'appwrite-worker'); $span->set('server.name', System::getEnv('_APP_LOGGING_SERVICE_IDENTIFIER', \gethostname())); $span->set('release', $version); $span->set('environment', System::getEnv('_APP_ENV', 'development') === 'production' ? 'production' : 'staging'); $span->set('appwrite.error.publish', true); $span->set('appwrite.error.action', $action); $span->set('verboseType', get_class($error)); $span->set('code', $error->getCode()); $span->set('projectId', $project->getId()); $span->set('roles', \json_encode($authorization->getRoles()) ?: null); $span->setError($error); $span->finish(); } finally { (new SpanCoroutineStorage())->set($activeSpan); } Console::error('[Error] Type: ' . get_class($error)); Console::error('[Error] Message: ' . $error->getMessage()); Console::error('[Error] File: ' . $error->getFile()); Console::error('[Error] Line: ' . $error->getLine()); }); $worker->start();