diff --git a/Dockerfile b/Dockerfile index 41592ef9fb..88d5ed030b 100755 --- a/Dockerfile +++ b/Dockerfile @@ -88,9 +88,7 @@ RUN chmod +x /usr/local/bin/doctor && \ chmod +x /usr/local/bin/worker-stats-usage && \ chmod +x /usr/local/bin/worker-stats-usage-dump && \ chmod +x /usr/local/bin/stats-resources && \ - chmod +x /usr/local/bin/worker-stats-resources && \ - chmod +x /usr/local/bin/worker-usage && \ - chmod +x /usr/local/bin/worker-usage-dump + chmod +x /usr/local/bin/worker-stats-resources # Letsencrypt Permissions RUN mkdir -p /etc/letsencrypt/live/ && chmod -Rf 755 /etc/letsencrypt/live/ diff --git a/app/controllers/api/functions.php b/app/controllers/api/functions.php index cdaf062788..583468f6c1 100644 --- a/app/controllers/api/functions.php +++ b/app/controllers/api/functions.php @@ -6,13 +6,14 @@ use Appwrite\Event\Build; use Appwrite\Event\Delete; use Appwrite\Event\Event; use Appwrite\Event\Func; +use Appwrite\Event\Realtime; use Appwrite\Event\StatsUsage; use Appwrite\Event\Validator\FunctionEvent; +use Appwrite\Event\Webhook; use Appwrite\Extend\Exception; use Appwrite\Extend\Exception as AppwriteException; use Appwrite\Functions\Validator\Headers; use Appwrite\Functions\Validator\RuntimeSpecification; -use Appwrite\Messaging\Adapter\Realtime; use Appwrite\Platform\Tasks\ScheduleExecutions; use Appwrite\SDK\AuthType; use Appwrite\SDK\ContentType; @@ -194,9 +195,12 @@ App::post('/v1/functions') ->inject('user') ->inject('queueForEvents') ->inject('queueForBuilds') + ->inject('queueForWebhooks') + ->inject('queueForFunctions') + ->inject('queueForRealtime') ->inject('dbForPlatform') ->inject('gitHub') - ->action(function (string $functionId, string $name, string $runtime, array $execute, array $events, string $schedule, int $timeout, bool $enabled, bool $logging, string $entrypoint, string $commands, array $scopes, string $installationId, string $providerRepositoryId, string $providerBranch, bool $providerSilentMode, string $providerRootDirectory, string $templateRepository, string $templateOwner, string $templateRootDirectory, string $templateVersion, string $specification, Request $request, Response $response, Database $dbForProject, callable $timelimit, Document $project, Document $user, Event $queueForEvents, Build $queueForBuilds, Database $dbForPlatform, GitHub $github) use ($redeployVcs) { + ->action(function (string $functionId, string $name, string $runtime, array $execute, array $events, string $schedule, int $timeout, bool $enabled, bool $logging, string $entrypoint, string $commands, array $scopes, string $installationId, string $providerRepositoryId, string $providerBranch, bool $providerSilentMode, string $providerRootDirectory, string $templateRepository, string $templateOwner, string $templateRootDirectory, string $templateVersion, string $specification, Request $request, Response $response, Database $dbForProject, callable $timelimit, Document $project, Document $user, Event $queueForEvents, Build $queueForBuilds, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Database $dbForPlatform, GitHub $github) use ($redeployVcs) { $functionId = ($functionId == 'unique()') ? ID::unique() : $functionId; // Temporary abuse check @@ -386,51 +390,29 @@ App::post('/v1/functions') ])) ); - /** Trigger Webhook */ $ruleModel = new Rule(); $ruleCreate = $queueForEvents - ->setClass(Event::WEBHOOK_CLASS_NAME) - ->setQueue(Event::WEBHOOK_QUEUE_NAME); + ->setProject($project) + ->setEvent('rules.[ruleId].create') + ->setParam('ruleId', $rule->getId()) + ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules()))); - $ruleCreate - ->setProject($project) - ->setEvent('rules.[ruleId].create') - ->setParam('ruleId', $rule->getId()) - ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules()))) + /** Trigger Webhook */ + $queueForWebhooks + ->from($ruleCreate) ->trigger(); /** Trigger Functions */ - $ruleCreate - ->setClass(Event::FUNCTIONS_CLASS_NAME) - ->setQueue(Event::FUNCTIONS_QUEUE_NAME) + $queueForFunctions + ->from($ruleCreate) ->trigger(); - /** Trigger realtime event */ - $allEvents = Event::generateEvents('rules.[ruleId].create', [ - 'ruleId' => $rule->getId(), - ]); - - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $allEvents[0], - payload: $rule, - project: $project - ); - Realtime::send( - projectId: 'console', - payload: $rule->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); - Realtime::send( - projectId: $project->getId(), - payload: $rule->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); + /** Trigger Realtime Events */ + $queueForRealtime + ->from($ruleCreate) + ->setSubscribers(['console', $project->getId()]) + ->trigger(); } $queueForEvents->setParam('functionId', $function->getId()); diff --git a/app/controllers/api/storage.php b/app/controllers/api/storage.php index bee245b64f..f13c9703c5 100644 --- a/app/controllers/api/storage.php +++ b/app/controllers/api/storage.php @@ -6,7 +6,6 @@ use Appwrite\Auth\Auth; use Appwrite\ClamAV\Network; use Appwrite\Event\Delete; use Appwrite\Event\Event; -use Appwrite\Event\StatsUsage; use Appwrite\Extend\Exception; use Appwrite\OpenSSL\OpenSSL; use Appwrite\SDK\AuthType; @@ -939,8 +938,7 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId/preview') ->inject('dbForProject') ->inject('deviceForFiles') ->inject('deviceForLocal') - ->inject('queueForStatsUsage') - ->action(function (string $bucketId, string $fileId, int $width, int $height, string $gravity, int $quality, int $borderWidth, string $borderColor, int $borderRadius, float $opacity, int $rotation, string $background, string $output, Response $response, Database $dbForProject, Device $deviceForFiles, Device $deviceForLocal, StatsUsage $queueForStatsUsage) { + ->action(function (string $bucketId, string $fileId, int $width, int $height, string $gravity, int $quality, int $borderWidth, string $borderColor, int $borderRadius, float $opacity, int $rotation, string $background, string $output, Response $response, Database $dbForProject, Device $deviceForFiles, Device $deviceForLocal) { if (!\extension_loaded('imagick')) { throw new Exception(Exception::GENERAL_SERVER_ERROR, 'Imagick extension is missing'); @@ -1068,14 +1066,13 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId/preview') $contentType = (\array_key_exists($output, $outputs)) ? $outputs[$output] : $outputs['jpg']; - $queueForStatsUsage - ->addMetric(METRIC_FILES_TRANSFORMATIONS, 1) - ->addMetric(str_replace('{bucketInternalId}', $bucket->getInternalId(), METRIC_BUCKET_ID_FILES_TRANSFORMATIONS), 1); - - $transformedAt = $file->getAttribute('transformedAt', ''); - if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) { - $file->setAttribute('transformedAt', DateTime::now()); - Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $file->getAttribute('bucketInternalId'), $file->getId(), $file)); + //Do not update transformedAt if it's a console user + if (!Auth::isPrivilegedUser(Authorization::getRoles())) { + $transformedAt = $file->getAttribute('transformedAt', ''); + if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) { + $file->setAttribute('transformedAt', DateTime::now()); + Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $file->getAttribute('bucketInternalId'), $file->getId(), $file)); + } } $response diff --git a/app/controllers/shared/api.php b/app/controllers/shared/api.php index 7f7b73ab0c..1015400a12 100644 --- a/app/controllers/shared/api.php +++ b/app/controllers/shared/api.php @@ -534,7 +534,7 @@ App::init() if ($type === 'bucket') { $bucketId = $parts[1] ?? null; - $bucket = Authorization::skip(fn () => $dbForProject->getDocument('buckets', $bucketId)); + $bucket = Authorization::skip(fn () => $dbForProject->getDocument('buckets', $bucketId)); if ($bucket->isEmpty() || (!$bucket->getAttribute('enabled') && !$isAppUser && !$isPrivilegedUser)) { throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND); @@ -560,11 +560,13 @@ App::init() if ($file->isEmpty()) { throw new Exception(Exception::STORAGE_FILE_NOT_FOUND); } - - $transformedAt = $file->getAttribute('transformedAt', ''); - if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) { - $file->setAttribute('transformedAt', DateTime::now()); - Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $file->getAttribute('bucketInternalId'), $file->getId(), $file)); + //Do not update transformedAt if it's a console user + if (!Auth::isPrivilegedUser(Authorization::getRoles())) { + $transformedAt = $file->getAttribute('transformedAt', ''); + if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) { + $file->setAttribute('transformedAt', DateTime::now()); + Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $file->getAttribute('bucketInternalId'), $file->getId(), $file)); + } } } diff --git a/app/init.php b/app/init.php index c0713ba327..1311a1429f 100644 --- a/app/init.php +++ b/app/init.php @@ -1200,9 +1200,6 @@ App::setResource('queueForAudits', function (Queue\Publisher $publisher) { App::setResource('queueForFunctions', function (Queue\Publisher $publisher) { return new Func($publisher); }, ['publisher']); -App::setResource('queueForUsage', function (Queue\Publisher $publisher) { - return new Usage($publisher); -}, ['publisher']); App::setResource('queueForCertificates', function (Queue\Publisher $publisher) { return new Certificate($publisher); }, ['publisher']); diff --git a/app/worker.php b/app/worker.php index 605474e9f1..f3e0541c9a 100644 --- a/app/worker.php +++ b/app/worker.php @@ -13,12 +13,12 @@ use Appwrite\Event\Func; use Appwrite\Event\Mail; use Appwrite\Event\Messaging; use Appwrite\Event\Migration; +use Appwrite\Event\Realtime; use Appwrite\Event\StatsUsage; use Appwrite\Event\StatsUsageDump; /** remove */ -use Appwrite\Event\Usage; -use Appwrite\Event\UsageDump; /** /remove */ +use Appwrite\Event\Webhook; use Appwrite\Platform\Appwrite; use Swoole\Runtime; use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis; @@ -269,14 +269,6 @@ Server::setResource('consumer', function (Group $pools) { return $pools->get('consumer')->pop()->getResource(); }, ['pools']); -Server::setResource('queueForUsage', function (Publisher $publisher) { - return new Usage($publisher); -}, ['publisher']); - -Server::setResource('queueForUsageDump', function (Publisher $publisher) { - return new UsageDump($publisher); -}, ['publisher']); - Server::setResource('queueForStatsUsage', function (Publisher $publisher) { return new StatsUsage($publisher); }, ['publisher']); @@ -313,10 +305,18 @@ Server::setResource('queueForAudits', function (Publisher $publisher) { return new Audit($publisher); }, ['publisher']); +Server::setResource('queueForWebhooks', function (Publisher $publisher) { + return new Webhook($publisher); +}, ['publisher']); + Server::setResource('queueForFunctions', function (Publisher $publisher) { return new Func($publisher); }, ['publisher']); +Server::setResource('queueForRealtime', function () { + return new Realtime(); +}, []); + Server::setResource('queueForCertificates', function (Publisher $publisher) { return new Certificate($publisher); }, ['publisher']); diff --git a/bin/worker-usage b/bin/worker-usage deleted file mode 100644 index e39ce8477c..0000000000 --- a/bin/worker-usage +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -php /usr/src/code/app/worker.php usage $@ \ No newline at end of file diff --git a/bin/worker-usage-dump b/bin/worker-usage-dump deleted file mode 100644 index 43ca87fcb3..0000000000 --- a/bin/worker-usage-dump +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -php /usr/src/code/app/worker.php usage-dump $@ \ No newline at end of file diff --git a/src/Appwrite/Event/Realtime.php b/src/Appwrite/Event/Realtime.php index f4f00b59d4..28a1bb6a6d 100644 --- a/src/Appwrite/Event/Realtime.php +++ b/src/Appwrite/Event/Realtime.php @@ -7,10 +7,17 @@ use Utopia\Database\Document; class Realtime extends Event { + protected array $subscribers = []; + public function __construct() { } + /** + * Get Realtime payload for this event. + * + * @return array + */ public function getRealtimePayload(): array { $payload = []; @@ -24,6 +31,28 @@ class Realtime extends Event return $payload; } + /** + * Set subscribers for this realtime event. + * + * @param array $subscribers + * @return array + */ + public function setSubscribers(array $subscribers): self + { + $this->subscribers = $subscribers; + return $this; + } + + /** + * Get subscribers for this realtime event. + * + * @return array + */ + public function getSubscribers(): array + { + return $this->subscribers; + } + /** * Execute Event. * @@ -53,17 +82,23 @@ class Realtime extends Event bucket: $bucket, ); - RealtimeAdapter::send( - projectId: $target['projectId'] ?? $this->getProject()->getId(), - payload: $this->getRealtimePayload(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'], - options: [ - 'permissionsChanged' => $target['permissionsChanged'], - 'userId' => $this->getParam('userId') - ] - ); + $projectIds = !empty($this->getSubscribers()) + ? $this->getSubscribers() + : [$target['projectId'] ?? $this->getProject()->getId()]; + + foreach ($projectIds as $projectId) { + RealtimeAdapter::send( + projectId: $projectId, + payload: $this->getRealtimePayload(), + events: $allEvents, + channels: $target['channels'], + roles: $target['roles'], + options: [ + 'permissionsChanged' => $target['permissionsChanged'], + 'userId' => $this->getParam('userId') + ] + ); + } return true; } diff --git a/src/Appwrite/Event/Usage.php b/src/Appwrite/Event/Usage.php deleted file mode 100644 index c70cea5c73..0000000000 --- a/src/Appwrite/Event/Usage.php +++ /dev/null @@ -1,78 +0,0 @@ -setQueue(Event::USAGE_QUEUE_NAME) - ->setClass(Event::USAGE_CLASS_NAME); - } - - /** - * Add reduce. - * - * @param Document $document - * @return self - */ - public function addReduce(Document $document): self - { - $this->reduce[] = $document; - - return $this; - } - - /** - * Add metric. - * - * @param string $key - * @param int $value - * @return self - */ - public function addMetric(string $key, int $value): self - { - - $this->metrics[] = [ - 'key' => $key, - 'value' => $value, - ]; - - return $this; - } - - /** - * Prepare the payload for the usage event. - * - * @return array - */ - protected function preparePayload(): array - { - return [ - 'project' => $this->project, - 'reduce' => $this->reduce, - 'metrics' => $this->metrics, - ]; - } - - /** - * Sends metrics to the usage worker. - * - * @return string|bool - */ - public function trigger(): string|bool - { - parent::trigger(); - $this->metrics = []; - return true; - } -} diff --git a/src/Appwrite/Event/UsageDump.php b/src/Appwrite/Event/UsageDump.php deleted file mode 100644 index a70716e94f..0000000000 --- a/src/Appwrite/Event/UsageDump.php +++ /dev/null @@ -1,44 +0,0 @@ -setQueue(Event::USAGE_DUMP_QUEUE_NAME) - ->setClass(Event::USAGE_DUMP_CLASS_NAME); - } - - /** - * Add Stats. - * - * @param array $stats - * @return self - */ - public function setStats(array $stats): self - { - $this->stats = $stats; - - return $this; - } - - /** - * Prepare the payload for the usage dump event. - * - * @return array - */ - protected function preparePayload(): array - { - return [ - 'stats' => $this->stats, - ]; - } -} diff --git a/src/Appwrite/Platform/Services/Workers.php b/src/Appwrite/Platform/Services/Workers.php index e121ee35f7..4f4095aca4 100644 --- a/src/Appwrite/Platform/Services/Workers.php +++ b/src/Appwrite/Platform/Services/Workers.php @@ -14,10 +14,6 @@ use Appwrite\Platform\Workers\Migrations; use Appwrite\Platform\Workers\StatsResources; use Appwrite\Platform\Workers\StatsUsage; use Appwrite\Platform\Workers\StatsUsageDump; -/** remove */ -use Appwrite\Platform\Workers\Usage; -use Appwrite\Platform\Workers\UsageDump; -/** /remove */ use Appwrite\Platform\Workers\Webhooks; use Utopia\Platform\Service; @@ -40,10 +36,6 @@ class Workers extends Service ->addAction(StatsUsage::getName(), new StatsUsage()) ->addAction(Migrations::getName(), new Migrations()) ->addAction(StatsResources::getName(), new StatsResources()) - /** Remove */ - ->addAction(UsageDump::getName(), new UsageDump()) - ->addAction(Usage::getName(), new Usage()) - /** /remove */ ; } } diff --git a/src/Appwrite/Platform/Tasks/StatsResources.php b/src/Appwrite/Platform/Tasks/StatsResources.php index c318f92cf1..ac3b9ead73 100644 --- a/src/Appwrite/Platform/Tasks/StatsResources.php +++ b/src/Appwrite/Platform/Tasks/StatsResources.php @@ -62,7 +62,7 @@ class StatsResources extends Action Authorization::disable(); Authorization::setDefaultStatus(false); - $last24Hours = (new \DateTime())->sub(\DateInterval::createFromDateString('3 hours')); + $last24Hours = (new \DateTime())->sub(\DateInterval::createFromDateString('24 hours')); /** * For each project that were accessed in last 24 hours */ diff --git a/src/Appwrite/Platform/Workers/Builds.php b/src/Appwrite/Platform/Workers/Builds.php index c9833adcfb..6f26e9a80c 100644 --- a/src/Appwrite/Platform/Workers/Builds.php +++ b/src/Appwrite/Platform/Workers/Builds.php @@ -5,8 +5,9 @@ namespace Appwrite\Platform\Workers; use Ahc\Jwt\JWT; use Appwrite\Event\Event; use Appwrite\Event\Func; +use Appwrite\Event\Realtime; use Appwrite\Event\StatsUsage; -use Appwrite\Messaging\Adapter\Realtime; +use Appwrite\Event\Webhook; use Appwrite\Utopia\Response\Model\Deployment; use Appwrite\Vcs\Comment; use Exception; @@ -49,15 +50,17 @@ class Builds extends Action ->inject('project') ->inject('dbForPlatform') ->inject('queueForEvents') + ->inject('queueForWebhooks') ->inject('queueForFunctions') + ->inject('queueForRealtime') ->inject('queueForStatsUsage') ->inject('cache') ->inject('dbForProject') ->inject('deviceForFunctions') ->inject('isResourceBlocked') ->inject('log') - ->callback(fn ($message, Document $project, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, StatsUsage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log) => - $this->action($message, $project, $dbForPlatform, $queueForEvents, $queueForFunctions, $usage, $cache, $dbForProject, $deviceForFunctions, $isResourceBlocked, $log)); + ->callback(fn ($message, Document $project, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, StatsUsage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log) => + $this->action($message, $project, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $usage, $cache, $dbForProject, $deviceForFunctions, $isResourceBlocked, $log)); } /** @@ -65,7 +68,9 @@ class Builds extends Action * @param Document $project * @param Database $dbForPlatform * @param Event $queueForEvents + * @param Webhook $queueForWebhooks * @param Func $queueForFunctions + * @param Realtime $queueForRealtime * @param StatsUsage $queueForStatsUsage * @param Cache $cache * @param Database $dbForProject @@ -74,7 +79,7 @@ class Builds extends Action * @return void * @throws \Utopia\Database\Exception */ - public function action(Message $message, Document $project, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, StatsUsage $queueForStatsUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log): void + public function action(Message $message, Document $project, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, StatsUsage $queueForStatsUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log): void { $payload = $message->getPayload() ?? []; @@ -95,7 +100,7 @@ class Builds extends Action case BUILD_TYPE_RETRY: Console::info('Creating build for deployment: ' . $deployment->getId()); $github = new GitHub($cache); - $this->buildDeployment($deviceForFunctions, $queueForFunctions, $queueForEvents, $queueForStatsUsage, $dbForPlatform, $dbForProject, $github, $project, $resource, $deployment, $template, $isResourceBlocked, $log); + $this->buildDeployment($deviceForFunctions, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $queueForEvents, $queueForStatsUsage, $dbForPlatform, $dbForProject, $github, $project, $resource, $deployment, $template, $isResourceBlocked, $log); break; default: @@ -105,7 +110,9 @@ class Builds extends Action /** * @param Device $deviceForFunctions + * @param Webhook $queueForWebhooks * @param Func $queueForFunctions + * @param Realtime $queueForRealtime * @param Event $queueForEvents * @param StatsUsage $queueForStatsUsage * @param Database $dbForPlatform @@ -120,7 +127,7 @@ class Builds extends Action * @throws \Utopia\Database\Exception * @throws Exception */ - protected function buildDeployment(Device $deviceForFunctions, Func $queueForFunctions, Event $queueForEvents, StatsUsage $queueForStatsUsage, Database $dbForPlatform, Database $dbForProject, GitHub $github, Document $project, Document $function, Document $deployment, Document $template, callable $isResourceBlocked, Log $log): void + protected function buildDeployment(Device $deviceForFunctions, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Event $queueForEvents, StatsUsage $queueForStatsUsage, Database $dbForPlatform, Database $dbForProject, GitHub $github, Document $project, Document $function, Document $deployment, Document $template, callable $isResourceBlocked, Log $log): void { $executor = new Executor(System::getEnv('_APP_EXECUTOR_HOST')); @@ -158,10 +165,7 @@ class Builds extends Action } // Realtime preparation - $allEvents = Event::generateEvents('functions.[functionId].deployments.[deploymentId].update', [ - 'functionId' => $function->getId(), - 'deploymentId' => $deployment->getId() - ]); + $event = "functions.[functionId].deployments.[deploymentId].update"; $startTime = DateTime::now(); $durationStart = \microtime(true); @@ -375,21 +379,16 @@ class Builds extends Action $deployment = $dbForProject->updateDocument('deployments', $deployment->getId(), $deployment); /** - * Send realtime Event + * Trigger Realtime Event */ - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $allEvents[0], - payload: $build, - project: $project - ); - Realtime::send( - projectId: 'console', - payload: $build->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); + $queueForRealtime + ->setProject($project) + ->setSubscribers(['console']) + ->setEvent($event) + ->setParam('functionId', $function->getId()) + ->setParam('deploymentId', $deployment->getId()) + ->setPayload($build->getArrayCopy()) + ->trigger(); } $tmpPath = '/tmp/builds/' . $buildId; @@ -436,40 +435,34 @@ class Builds extends Action $this->runGitAction('building', $github, $providerCommitHash, $owner, $repositoryName, $project, $function, $deployment->getId(), $dbForProject, $dbForPlatform); } - /** Trigger Webhook */ $deploymentModel = new Deployment(); $deploymentUpdate = $queueForEvents - ->setQueue(Event::WEBHOOK_QUEUE_NAME) - ->setClass(Event::WEBHOOK_CLASS_NAME) ->setProject($project) ->setEvent('functions.[functionId].deployments.[deploymentId].update') ->setParam('functionId', $function->getId()) ->setParam('deploymentId', $deployment->getId()) ->setPayload($deployment->getArrayCopy(array_keys($deploymentModel->getRules()))); - $deploymentUpdate->trigger(); + /** Trigger Webhook */ + $queueForWebhooks + ->from($deploymentUpdate) + ->trigger(); /** Trigger Functions */ $queueForFunctions ->from($deploymentUpdate) ->trigger(); - /** Trigger Realtime */ - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $allEvents[0], - payload: $build, - project: $project - ); - - Realtime::send( - projectId: 'console', - payload: $build->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); + /** Trigger Realtime Event */ + $queueForRealtime + ->setProject($project) + ->setSubscribers(['console']) + ->setEvent($event) + ->setParam('functionId', $function->getId()) + ->setParam('deploymentId', $deployment->getId()) + ->setPayload($build->getArrayCopy()) + ->trigger(); $vars = []; @@ -562,12 +555,12 @@ class Builds extends Action $err = $error; } }), - Co\go(function () use ($executor, $project, $deployment, &$response, &$build, $dbForProject, $allEvents, &$err, &$isCanceled) { + Co\go(function () use ($executor, $project, $function, $deployment, &$response, &$build, $dbForProject, $event, &$err, $queueForRealtime, &$isCanceled) { try { $executor->getLogs( deploymentId: $deployment->getId(), projectId: $project->getId(), - callback: function ($logs) use (&$response, &$err, &$build, $dbForProject, $allEvents, $project, &$isCanceled) { + callback: function ($logs) use (&$response, &$err, &$build, $dbForProject, $event, $project, $function, $deployment, $queueForRealtime, &$isCanceled) { if ($isCanceled) { return; } @@ -592,21 +585,16 @@ class Builds extends Action $build = $dbForProject->updateDocument('builds', $build->getId(), $build); /** - * Send realtime Event + * Trigger Realtime Event */ - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $allEvents[0], - payload: $build, - project: $project - ); - Realtime::send( - projectId: 'console', - payload: $build->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); + $queueForRealtime + ->setProject($project) + ->setSubscribers(['console']) + ->setEvent($event) + ->setParam('functionId', $function->getId()) + ->setParam('deploymentId', $deployment->getId()) + ->setPayload($build->getArrayCopy()) + ->trigger(); } } ); @@ -694,21 +682,16 @@ class Builds extends Action } } finally { /** - * Send realtime Event + * Trigger Realtime Event */ - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $allEvents[0], - payload: $build, - project: $project - ); - Realtime::send( - projectId: 'console', - payload: $build->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); + $queueForRealtime + ->setProject($project) + ->setSubscribers(['console']) + ->setEvent($event) + ->setParam('functionId', $function->getId()) + ->setParam('deploymentId', $deployment->getId()) + ->setPayload($build->getArrayCopy()) + ->trigger(); /** Trigger usage queue */ if ($build->getAttribute('status') === 'ready') { diff --git a/src/Appwrite/Platform/Workers/Certificates.php b/src/Appwrite/Platform/Workers/Certificates.php index 7e220b2734..093e6fda5a 100644 --- a/src/Appwrite/Platform/Workers/Certificates.php +++ b/src/Appwrite/Platform/Workers/Certificates.php @@ -6,7 +6,8 @@ use Appwrite\Certificates\Adapter as CertificatesAdapter; use Appwrite\Event\Event; use Appwrite\Event\Func; use Appwrite\Event\Mail; -use Appwrite\Messaging\Adapter\Realtime; +use Appwrite\Event\Realtime; +use Appwrite\Event\Webhook; use Appwrite\Network\Validator\CNAME; use Appwrite\Template\Template; use Appwrite\Utopia\Response\Model\Rule; @@ -46,12 +47,14 @@ class Certificates extends Action ->inject('dbForPlatform') ->inject('queueForMails') ->inject('queueForEvents') + ->inject('queueForWebhooks') ->inject('queueForFunctions') + ->inject('queueForRealtime') ->inject('log') ->inject('certificates') ->callback( - fn (Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates) => - $this->action($message, $dbForPlatform, $queueForMails, $queueForEvents, $queueForFunctions, $log, $certificates) + fn (Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates) => + $this->action($message, $dbForPlatform, $queueForMails, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $log, $certificates) ); } @@ -60,14 +63,16 @@ class Certificates extends Action * @param Database $dbForPlatform * @param Mail $queueForMails * @param Event $queueForEvents + * @param Webhook $queueForWebhooks * @param Func $queueForFunctions + * @param Realtime $queueForRealtime * @param Log $log * @param CertificatesAdapter $certificates * @return void * @throws Throwable * @throws \Utopia\Database\Exception */ - public function action(Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates): void + public function action(Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates): void { $payload = $message->getPayload() ?? []; @@ -81,7 +86,7 @@ class Certificates extends Action $log->addTag('domain', $domain->get()); - $this->execute($domain, $dbForPlatform, $queueForMails, $queueForEvents, $queueForFunctions, $log, $certificates, $skipRenewCheck); + $this->execute($domain, $dbForPlatform, $queueForMails, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $log, $certificates, $skipRenewCheck); } /** @@ -90,13 +95,14 @@ class Certificates extends Action * @param Mail $queueForMails * @param Event $queueForEvents * @param Func $queueForFunctions + * @param Realtime $queueForRealtime * @param CertificatesAdapter $certificates * @param bool $skipRenewCheck * @return void * @throws Throwable * @throws \Utopia\Database\Exception */ - private function execute(Domain $domain, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates, bool $skipRenewCheck = false): void + private function execute(Domain $domain, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates, bool $skipRenewCheck = false): void { /** * 1. Read arguments and validate domain @@ -186,7 +192,7 @@ class Certificates extends Action $certificate->setAttribute('updated', DateTime::now()); // Save all changes we made to certificate document into database - $this->saveCertificateDocument($domain->get(), $certificate, $success, $dbForPlatform, $queueForEvents, $queueForFunctions); + $this->saveCertificateDocument($domain->get(), $certificate, $success, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime); } } @@ -199,13 +205,14 @@ class Certificates extends Action * @param Database $dbForPlatform Database connection for console * @param Event $queueForEvents * @param Func $queueForFunctions + * @param Realtime $queueForRealtime * @return void * @throws \Utopia\Database\Exception * @throws Authorization * @throws Conflict * @throws Structure */ - private function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions): void + private function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime): void { // Check if update or insert required $certificateDocument = $dbForPlatform->findOne('certificates', [Query::equal('domain', [$domain])]); @@ -219,7 +226,7 @@ class Certificates extends Action } $certificateId = $certificate->getId(); - $this->updateDomainDocuments($certificateId, $domain, $success, $dbForPlatform, $queueForEvents, $queueForFunctions); + $this->updateDomainDocuments($certificateId, $domain, $success, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime); } /** @@ -338,7 +345,7 @@ class Certificates extends Action * * @return void */ - private function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions): void + private function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime): void { // TODO: @christyjacob remove once we migrate the rules in 1.7.x if (System::getEnv('_APP_RULES_FORMAT') === 'md5') { @@ -367,50 +374,28 @@ class Certificates extends Action return; } - /** Trigger Webhook */ $ruleModel = new Rule(); $queueForEvents - ->setQueue(Event::WEBHOOK_QUEUE_NAME) - ->setClass(Event::WEBHOOK_CLASS_NAME) ->setProject($project) ->setEvent('rules.[ruleId].update') ->setParam('ruleId', $rule->getId()) - ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules()))) - ->trigger(); + ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules()))); + /** Trigger Webhook */ + $queueForWebhooks + ->from($queueForEvents) + ->trigger(); /** Trigger Functions */ $queueForFunctions - ->setProject($project) - ->setEvent('rules.[ruleId].update') - ->setParam('ruleId', $rule->getId()) - ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules()))) + ->from($queueForEvents) ->trigger(); - /** Trigger realtime event */ - $allEvents = Event::generateEvents('rules.[ruleId].update', [ - 'ruleId' => $rule->getId(), - ]); - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $allEvents[0], - payload: $rule, - project: $project - ); - Realtime::send( - projectId: 'console', - payload: $rule->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); - Realtime::send( - projectId: $project->getId(), - payload: $rule->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); + /** Trigger Realtime Events */ + $queueForRealtime + ->from($queueForEvents) + ->setSubscribers(['console', $projectId]) + ->trigger(); } } } diff --git a/src/Appwrite/Platform/Workers/Databases.php b/src/Appwrite/Platform/Workers/Databases.php index 0f400e0107..4abd035599 100644 --- a/src/Appwrite/Platform/Workers/Databases.php +++ b/src/Appwrite/Platform/Workers/Databases.php @@ -2,8 +2,7 @@ namespace Appwrite\Platform\Workers; -use Appwrite\Event\Event; -use Appwrite\Messaging\Adapter\Realtime; +use Appwrite\Event\Realtime; use Exception; use Utopia\CLI\Console; use Utopia\Database\Database; @@ -37,8 +36,9 @@ class Databases extends Action ->inject('project') ->inject('dbForPlatform') ->inject('dbForProject') + ->inject('queueForRealtime') ->inject('log') - ->callback(fn (Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Log $log) => $this->action($message, $project, $dbForPlatform, $dbForProject, $log)); + ->callback(fn (Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime, Log $log) => $this->action($message, $project, $dbForPlatform, $dbForProject, $queueForRealtime, $log)); } /** @@ -46,16 +46,17 @@ class Databases extends Action * @param Document $project * @param Database $dbForPlatform * @param Database $dbForProject + * @param Realtime $queueForRealtime * @param Log $log * @return void * @throws \Exception */ - public function action(Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Log $log): void + public function action(Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime, Log $log): void { $payload = $message->getPayload() ?? []; if (empty($payload)) { - throw new \Exception('Missing payload'); + throw new Exception('Missing payload'); } $type = $payload['type']; @@ -75,11 +76,11 @@ class Databases extends Action match (\strval($type)) { DATABASE_TYPE_DELETE_DATABASE => $this->deleteDatabase($database, $project, $dbForProject), DATABASE_TYPE_DELETE_COLLECTION => $this->deleteCollection($database, $collection, $project, $dbForProject), - DATABASE_TYPE_CREATE_ATTRIBUTE => $this->createAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject), - DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject), - DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject), - DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject), - default => throw new \Exception('No database operation for type: ' . \strval($type)), + DATABASE_TYPE_CREATE_ATTRIBUTE => $this->createAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime), + DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime), + DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime), + DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime), + default => throw new Exception('No database operation for type: ' . \strval($type)), }; } @@ -90,13 +91,14 @@ class Databases extends Action * @param Document $project * @param Database $dbForPlatform * @param Database $dbForProject + * @param Realtime $queueForRealtime * @return void * @throws Authorization * @throws Conflict * @throws \Exception * @throws \Throwable */ - private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject): void + private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void { if ($collection->isEmpty()) { throw new Exception('Missing collection'); @@ -106,12 +108,7 @@ class Databases extends Action } $projectId = $project->getId(); - - $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].attributes.[attributeId].update', [ - 'databaseId' => $database->getId(), - 'collectionId' => $collection->getId(), - 'attributeId' => $attribute->getId() - ]); + $event = "databases.[databaseId].collections.[collectionId].attributes.[attributeId].update"; /** * TODO @christyjacob4 verify if this is still the case * Fetch attribute from the database, since with Resque float values are loosing informations. @@ -169,7 +166,7 @@ class Databases extends Action break; default: if (!$dbForProject->createAttribute('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $key, $type, $size, $required, $default, $signed, $array, $format, $formatOptions, $filters)) { - throw new \Exception('Failed to create Attribute'); + throw new Exception('Failed to create Attribute'); } } @@ -200,7 +197,7 @@ class Databases extends Action throw $e; } finally { - $this->trigger($database, $collection, $attribute, $project, $projectId, $events); + $this->trigger($database, $collection, $project, $event, $queueForRealtime, $attribute); if (! $relatedCollection->isEmpty()) { $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $relatedCollection->getId()); @@ -217,13 +214,14 @@ class Databases extends Action * @param Document $project * @param Database $dbForPlatform * @param Database $dbForProject + * @param Realtime $queueForRealtime * @return void * @throws Authorization * @throws Conflict * @throws \Exception * @throws \Throwable **/ - private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject): void + private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void { if ($collection->isEmpty()) { throw new Exception('Missing collection'); @@ -233,15 +231,9 @@ class Databases extends Action } $projectId = $project->getId(); - - $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].attributes.[attributeId].delete', [ - 'databaseId' => $database->getId(), - 'collectionId' => $collection->getId(), - 'attributeId' => $attribute->getId() - ]); + $event = 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].delete'; $collectionId = $collection->getId(); $key = $attribute->getAttribute('key', ''); - $status = $attribute->getAttribute('status', ''); $type = $attribute->getAttribute('type', ''); $project = $dbForPlatform->getDocument('projects', $projectId); $options = $attribute->getAttribute('options', []); @@ -312,7 +304,7 @@ class Databases extends Action throw $e; } finally { - $this->trigger($database, $collection, $attribute, $project, $projectId, $events); + $this->trigger($database, $collection, $project, $event, $queueForRealtime, $attribute); } // The underlying database removes/rebuilds indexes when attribute is removed @@ -358,7 +350,7 @@ class Databases extends Action } if ($exists) { // Delete the duplicate if created, else update in db - $this->deleteIndex($database, $collection, $index, $project, $dbForPlatform, $dbForProject); + $this->deleteIndex($database, $collection, $index, $project, $dbForPlatform, $dbForProject, $queueForRealtime); } else { $dbForProject->updateDocument('indexes', $index->getId(), $index); } @@ -381,6 +373,7 @@ class Databases extends Action * @param Document $project * @param Database $dbForPlatform * @param Database $dbForProject + * @param Realtime $queueForRealtime * @return void * @throws Authorization * @throws Conflict @@ -388,7 +381,7 @@ class Databases extends Action * @throws DatabaseException * @throws \Throwable */ - private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject): void + private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void { if ($collection->isEmpty()) { throw new Exception('Missing collection'); @@ -398,12 +391,7 @@ class Databases extends Action } $projectId = $project->getId(); - - $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].indexes.[indexId].update', [ - 'databaseId' => $database->getId(), - 'collectionId' => $collection->getId(), - 'indexId' => $index->getId() - ]); + $event = 'databases.[databaseId].collections.[collectionId].indexes.[indexId].update'; $collectionId = $collection->getId(); $key = $index->getAttribute('key', ''); $type = $index->getAttribute('type', ''); @@ -430,7 +418,7 @@ class Databases extends Action throw $e; } finally { - $this->trigger($database, $collection, $index, $project, $projectId, $events); + $this->trigger($database, $collection, $project, $event, $queueForRealtime, null, $index); $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collectionId); } } @@ -442,6 +430,7 @@ class Databases extends Action * @param Document $project * @param Database $dbForPlatform * @param Database $dbForProject + * @param Realtime $queueForRealtime * @return void * @throws Authorization * @throws Conflict @@ -449,7 +438,7 @@ class Databases extends Action * @throws DatabaseException * @throws \Throwable */ - private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject): void + private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void { if ($collection->isEmpty()) { throw new Exception('Missing collection'); @@ -459,12 +448,7 @@ class Databases extends Action } $projectId = $project->getId(); - - $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].indexes.[indexId].delete', [ - 'databaseId' => $database->getId(), - 'collectionId' => $collection->getId(), - 'indexId' => $index->getId() - ]); + $event = 'databases.[databaseId].collections.[collectionId].indexes.[indexId].delete'; $key = $index->getAttribute('key'); $status = $index->getAttribute('status', ''); $project = $dbForPlatform->getDocument('projects', $projectId); @@ -490,7 +474,7 @@ class Databases extends Action throw $e; } finally { - $this->trigger($database, $collection, $index, $project, $projectId, $events); + $this->trigger($database, $collection, $project, $event, $queueForRealtime, null, $index); $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collection->getId()); } } @@ -532,7 +516,6 @@ class Databases extends Action $collectionId = $collection->getId(); $collectionInternalId = $collection->getInternalId(); - $databaseId = $database->getId(); $databaseInternalId = $database->getInternalId(); $dbForProject->deleteCollection('database_' . $databaseInternalId . '_collection_' . $collection->getInternalId()); @@ -568,21 +551,21 @@ class Databases extends Action /** - * @param string $collection collectionID + * @param string $collectionId * @param array $queries * @param Database $database * @param callable|null $callback * @return void * @throws Exception */ - protected function deleteByGroup(string $collection, array $queries, Database $database, callable $callback = null): void + protected function deleteByGroup(string $collectionId, array $queries, Database $database, callable $callback = null): void { $start = \microtime(true); try { - $documents = $database->deleteDocuments($collection, $queries); + $documents = $database->deleteDocuments($collectionId, $queries); } catch (\Throwable $th) { - Console::error('Failed to delete documents for collection ' . $collection . ': ' . $th->getMessage()); + Console::error('Failed to delete documents for collection ' . $collectionId . ': ' . $th->getMessage()); return; } @@ -598,31 +581,42 @@ class Databases extends Action Console::info("Deleted {$count} documents by group in " . ($end - $start) . " seconds"); } + /** + * @param Document $database + * @param Document $collection + * @param Document $project + * @param Realtime $queueForRealtime + * @param Document|null $attribute + * @param Document|null $index + * @return void + */ protected function trigger( Document $database, Document $collection, - Document $attribute, Document $project, - string $projectId, - array $events + string $event, + Realtime $queueForRealtime, + Document|null $attribute = null, + Document|null $index = null, ): void { - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $events[0], - payload: $attribute, - project: $project, - ); - Realtime::send( - projectId: 'console', - payload: $attribute->getArrayCopy(), - events: $events, - channels: $target['channels'], - roles: $target['roles'], - options: [ - 'projectId' => $projectId, - 'databaseId' => $database->getId(), - 'collectionId' => $collection->getId() - ] - ); + $queueForRealtime + ->setProject($project) + ->setSubscribers(['console']) + ->setEvent($event) + ->setParam('databaseId', $database->getId()) + ->setParam('collectionId', $collection->getId()); + + if ($attribute !== null && !empty($attribute)) { + $queueForRealtime + ->setParam('attributeId', $attribute->getId()) + ->setPayload($attribute->getArrayCopy()); + } + if ($index !== null && !empty($index)) { + $queueForRealtime + ->setParam('indexId', $index->getId()) + ->setPayload($index->getArrayCopy()); + } + + $queueForRealtime->trigger(); } } diff --git a/src/Appwrite/Platform/Workers/Functions.php b/src/Appwrite/Platform/Workers/Functions.php index 0a7c39c02f..a7caa3207f 100644 --- a/src/Appwrite/Platform/Workers/Functions.php +++ b/src/Appwrite/Platform/Workers/Functions.php @@ -5,8 +5,9 @@ namespace Appwrite\Platform\Workers; use Ahc\Jwt\JWT; use Appwrite\Event\Event; use Appwrite\Event\Func; +use Appwrite\Event\Realtime; use Appwrite\Event\StatsUsage; -use Appwrite\Messaging\Adapter\Realtime; +use Appwrite\Event\Webhook; use Appwrite\Utopia\Response\Model\Execution; use Exception; use Executor\Executor; @@ -44,15 +45,17 @@ class Functions extends Action ->inject('project') ->inject('message') ->inject('dbForProject') + ->inject('queueForWebhooks') ->inject('queueForFunctions') + ->inject('queueForRealtime') ->inject('queueForEvents') ->inject('queueForStatsUsage') ->inject('log') ->inject('isResourceBlocked') - ->callback(fn (Document $project, Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked) => $this->action($project, $message, $dbForProject, $queueForFunctions, $queueForEvents, $queueForStatsUsage, $log, $isResourceBlocked)); + ->callback(fn (Document $project, Message $message, Database $dbForProject, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked) => $this->action($project, $message, $dbForProject, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $queueForEvents, $queueForStatsUsage, $log, $isResourceBlocked)); } - public function action(Document $project, Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked): void + public function action(Document $project, Message $message, Database $dbForProject, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked): void { $payload = $message->getPayload() ?? []; @@ -136,7 +139,9 @@ class Functions extends Action $this->execute( log: $log, dbForProject: $dbForProject, + queueForWebhooks: $queueForWebhooks, queueForFunctions: $queueForFunctions, + queueForRealtime: $queueForRealtime, queueForStatsUsage: $queueForStatsUsage, queueForEvents: $queueForEvents, project: $project, @@ -176,7 +181,9 @@ class Functions extends Action $this->execute( log: $log, dbForProject: $dbForProject, + queueForWebhooks: $queueForWebhooks, queueForFunctions: $queueForFunctions, + queueForRealtime: $queueForRealtime, queueForStatsUsage: $queueForStatsUsage, queueForEvents: $queueForEvents, project: $project, @@ -198,7 +205,9 @@ class Functions extends Action $this->execute( log: $log, dbForProject: $dbForProject, + queueForWebhooks: $queueForWebhooks, queueForFunctions: $queueForFunctions, + queueForRealtime: $queueForRealtime, queueForStatsUsage: $queueForStatsUsage, queueForEvents: $queueForEvents, project: $project, @@ -284,6 +293,7 @@ class Functions extends Action * @param Log $log * @param Database $dbForProject * @param Func $queueForFunctions + * @param Realtime $queueForRealtime * @param StatsUsage $queueForStatsUsage * @param Event $queueForEvents * @param Document $project @@ -307,7 +317,9 @@ class Functions extends Action private function execute( Log $log, Database $dbForProject, + Webhook $queueForWebhooks, Func $queueForFunctions, + Realtime $queueForRealtime, StatsUsage $queueForStatsUsage, Event $queueForEvents, Document $project, @@ -564,20 +576,20 @@ class Functions extends Action ; } - $execution = $dbForProject->updateDocument('executions', $executionId, $execution); - /** Trigger Webhook */ $executionModel = new Execution(); $queueForEvents - ->setQueue(Event::WEBHOOK_QUEUE_NAME) - ->setClass(Event::WEBHOOK_CLASS_NAME) ->setProject($project) ->setUser($user) ->setEvent('functions.[functionId].executions.[executionId].update') ->setParam('functionId', $function->getId()) ->setParam('executionId', $execution->getId()) - ->setPayload($execution->getArrayCopy(array_keys($executionModel->getRules()))) + ->setPayload($execution->getArrayCopy(array_keys($executionModel->getRules()))); + + /** Trigger Webhook */ + $queueForWebhooks + ->from($queueForEvents) ->trigger(); /** Trigger Functions */ @@ -585,31 +597,11 @@ class Functions extends Action ->from($queueForEvents) ->trigger(); - /** Trigger realtime event */ - $allEvents = Event::generateEvents('functions.[functionId].executions.[executionId].update', [ - 'functionId' => $function->getId(), - 'executionId' => $execution->getId() - ]); - $target = Realtime::fromPayload( - // Pass first, most verbose event pattern - event: $allEvents[0], - payload: $execution, - project: $project - ); - Realtime::send( - projectId: 'console', - payload: $execution->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); - Realtime::send( - projectId: $project->getId(), - payload: $execution->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'] - ); + /** Trigger Realtime Events */ + $queueForRealtime + ->from($queueForEvents) + ->setSubscribers(['console', $project->getId()]) + ->trigger(); if (!empty($error)) { throw new Exception($error, $errorCode); diff --git a/src/Appwrite/Platform/Workers/Migrations.php b/src/Appwrite/Platform/Workers/Migrations.php index 3bf8e06356..f21a846a0d 100644 --- a/src/Appwrite/Platform/Workers/Migrations.php +++ b/src/Appwrite/Platform/Workers/Migrations.php @@ -3,8 +3,7 @@ namespace Appwrite\Platform\Workers; use Ahc\Jwt\JWT; -use Appwrite\Event\Event; -use Appwrite\Messaging\Adapter\Realtime; +use Appwrite\Event\Realtime; use Exception; use Utopia\CLI\Console; use Utopia\Config\Config; @@ -54,13 +53,14 @@ class Migrations extends Action ->inject('dbForProject') ->inject('dbForPlatform') ->inject('logError') - ->callback(fn (Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError) => $this->action($message, $project, $dbForProject, $dbForPlatform, $logError)); + ->inject('queueForRealtime') + ->callback(fn (Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime) => $this->action($message, $project, $dbForProject, $dbForPlatform, $logError, $queueForRealtime)); } /** * @throws Exception */ - public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError): void + public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime): void { $payload = $message->getPayload() ?? []; @@ -87,7 +87,7 @@ class Migrations extends Action return; } - $this->processMigration($migration); + $this->processMigration($migration, $queueForRealtime); } /** @@ -155,34 +155,16 @@ class Migrations extends Action * @throws \Utopia\Database\Exception * @throws Exception */ - protected function updateMigrationDocument(Document $migration, Document $project): Document + protected function updateMigrationDocument(Document $migration, Document $project, Realtime $queueForRealtime): Document { - /** Trigger Realtime */ - $allEvents = Event::generateEvents('migrations.[migrationId].update', [ - 'migrationId' => $migration->getId(), - ]); - - $target = Realtime::fromPayload( - event: $allEvents[0], - payload: $migration, - project: $project - ); - - Realtime::send( - projectId: 'console', - payload: $migration->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'], - ); - - Realtime::send( - projectId: $project->getId(), - payload: $migration->getArrayCopy(), - events: $allEvents, - channels: $target['channels'], - roles: $target['roles'], - ); + /** Trigger Realtime Events */ + $queueForRealtime + ->setProject($project) + ->setSubscribers(['console', $project->getId()]) + ->setEvent('migrations.[migrationId].update') + ->setParam('migrationId', $migration->getId()) + ->setPayload($migration->getArrayCopy()) + ->trigger(); return $this->dbForProject->updateDocument('migrations', $migration->getId(), $migration); } @@ -231,7 +213,7 @@ class Migrations extends Action * @throws \Utopia\Database\Exception * @throws Exception */ - protected function processMigration(Document $migration): void + protected function processMigration(Document $migration, Realtime $queueForRealtime): void { $project = $this->project; $projectDocument = $this->dbForPlatform->getDocument('projects', $project->getId()); @@ -255,7 +237,7 @@ class Migrations extends Action $migration->setAttribute('stage', 'processing'); $migration->setAttribute('status', 'processing'); - $this->updateMigrationDocument($migration, $projectDocument); + $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime); $source = $this->processSource($migration); $destination = $this->processDestination($migration, $tempAPIKey); @@ -269,14 +251,14 @@ class Migrations extends Action /** Start Transfer */ $migration->setAttribute('stage', 'migrating'); - $this->updateMigrationDocument($migration, $projectDocument); + $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime); $transfer->run( $migration->getAttribute('resources'), - function () use ($migration, $transfer, $projectDocument) { + function () use ($migration, $transfer, $projectDocument, $queueForRealtime) { $migration->setAttribute('resourceData', json_encode($transfer->getCache())); $migration->setAttribute('statusCounters', json_encode($transfer->getStatusCounters())); - $this->updateMigrationDocument($migration, $projectDocument); + $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime); }, $migration->getAttribute('resourceId'), $migration->getAttribute('resourceType') @@ -313,7 +295,7 @@ class Migrations extends Action } $migration->setAttribute('errors', $errorMessages); - $this->updateMigrationDocument($migration, $projectDocument); + $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime); return; } @@ -354,7 +336,7 @@ class Migrations extends Action $migration->setAttribute('errors', $errorMessages); } } finally { - $this->updateMigrationDocument($migration, $projectDocument); + $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime); if ($migration->getAttribute('status', '') === 'failed') { Console::error('Migration('.$migration->getInternalId().':'.$migration->getId().') failed, Project('.$this->project->getInternalId().':'.$this->project->getId().')'); diff --git a/src/Appwrite/Platform/Workers/StatsUsageDump.php b/src/Appwrite/Platform/Workers/StatsUsageDump.php index 5d7240f4b5..81acd4e4b0 100644 --- a/src/Appwrite/Platform/Workers/StatsUsageDump.php +++ b/src/Appwrite/Platform/Workers/StatsUsageDump.php @@ -117,7 +117,7 @@ class StatsUsageDump extends Action $project = new Document($stats['project'] ?? []); $numberOfKeys = !empty($stats['keys']) ? count($stats['keys']) : 0; - $receivedAt = $stats['receivedAt'] ?? 'NONE'; + $receivedAt = $stats['receivedAt'] ?? null; if ($numberOfKeys === 0) { continue; } @@ -134,7 +134,7 @@ class StatsUsageDump extends Action if (str_contains($key, METRIC_DATABASES_STORAGE)) { try { - $this->handleDatabaseStorage($key, $dbForProject, $project); + $this->handleDatabaseStorage($key, $dbForProject, $project, $receivedAt); } catch (\Exception $e) { console::error('[' . DateTime::now() . '] failed to calculate database storage for key [' . $key . '] ' . $e->getMessage()); } @@ -142,7 +142,11 @@ class StatsUsageDump extends Action } foreach ($this->periods as $period => $format) { - $time = 'inf' === $period ? null : date($format, time()); + $time = null; + + if ($period !== 'inf') { + $time = !empty($receivedAt) ? (new \DateTime($receivedAt))->format($format) : date($format, time()); + } $id = \md5("{$time}_{$period}_{$key}"); $document = new Document([ @@ -171,12 +175,12 @@ class StatsUsageDump extends Action } } - private function handleDatabaseStorage(string $key, Database $dbForProject, Document $project): void + private function handleDatabaseStorage(string $key, Database $dbForProject, Document $project, string $receivedAt): void { $data = explode('.', $key); $start = microtime(true); - $updateMetric = function (Database $dbForProject, Document $project, int $value, string $key, string $period, string|null $time) { + $updateMetric = function (Database $dbForProject, Document $project, int $value, string $key, string $period, string|null $time) use ($receivedAt) { $id = \md5("{$time}_{$period}_{$key}"); $document = new Document([ @@ -197,7 +201,11 @@ class StatsUsageDump extends Action }; foreach ($this->periods as $period => $format) { - $time = 'inf' === $period ? null : date($format, time()); + $time = null; + + if ($period !== 'inf') { + $time = !empty($receivedAt) ? (new \DateTime($receivedAt))->format($format) : date($format, time()); + } $id = \md5("{$time}_{$period}_{$key}"); $value = 0; @@ -320,7 +328,7 @@ class StatsUsageDump extends Action protected function writeToLogsDB(Document $project, Document $document): void { - if (!System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', false)) { + if (System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', 'disabled') === 'disabled') { Console::log('Dual Writing is disabled. Skipping...'); return; } diff --git a/src/Appwrite/Platform/Workers/Usage.php b/src/Appwrite/Platform/Workers/Usage.php deleted file mode 100644 index 3687eeab67..0000000000 --- a/src/Appwrite/Platform/Workers/Usage.php +++ /dev/null @@ -1,292 +0,0 @@ -desc('Usage worker') - ->inject('message') - ->inject('project') - ->inject('getProjectDB') - ->inject('queueForUsageDump') - ->callback(function (Message $message, Document $project, callable $getProjectDB, UsageDump $queueForUsageDump) { - $this->action($message, $project, $getProjectDB, $queueForUsageDump); - }); - - $this->aggregationInterval = (int) System::getEnv('_APP_USAGE_AGGREGATION_INTERVAL', '20'); - $this->lastTriggeredTime = time(); - } - - /** - * @param Message $message - * @param Document $project - * @param callable $getProjectDB - * @param UsageDump $queueForUsageDump - * @return void - * @throws \Utopia\Database\Exception - * @throws Exception - */ - public function action(Message $message, Document $project, callable $getProjectDB, UsageDump $queueForUsageDump): void - { - $payload = $message->getPayload() ?? []; - if (empty($payload)) { - throw new Exception('Missing payload'); - } - - - if (empty($project->getAttribute('database'))) { - var_dump($payload); - return; - } - - $projectId = $project->getInternalId(); - foreach ($payload['reduce'] ?? [] as $document) { - if (empty($document)) { - continue; - } - - $this->reduce( - project: $project, - document: new Document($document), - metrics: $payload['metrics'], - getProjectDB: $getProjectDB - ); - } - - - $this->stats[$projectId]['project'] = [ - '$id' => $project->getId(), - '$internalId' => $project->getInternalId(), - 'database' => $project->getAttribute('database'), - ]; - $this->stats[$projectId]['receivedAt'] = DateTime::now(); - foreach ($payload['metrics'] ?? [] as $metric) { - $this->keys++; - if (!isset($this->stats[$projectId]['keys'][$metric['key']])) { - $this->stats[$projectId]['keys'][$metric['key']] = $metric['value']; - continue; - } - - $this->stats[$projectId]['keys'][$metric['key']] += $metric['value']; - } - - // If keys crossed threshold or X time passed since the last send and there are some keys in the array ($this->stats) - if ( - $this->keys >= self::KEYS_THRESHOLD || - (time() - $this->lastTriggeredTime > $this->aggregationInterval && $this->keys > 0) - ) { - Console::warning('[' . DateTime::now() . '] Aggregated ' . $this->keys . ' keys'); - - $queueForUsageDump - ->setStats($this->stats) - ->trigger(); - - $this->stats = []; - $this->keys = 0; - $this->lastTriggeredTime = time(); - } - } - - /** - * On Documents that tied by relations like functions>deployments>build || documents>collection>database || buckets>files. - * When we remove a parent document we need to deduct his children aggregation from the project scope. - * @param Document $project - * @param Document $document - * @param array $metrics - * @param callable $getProjectDB - * @return void - */ - private function reduce(Document $project, Document $document, array &$metrics, callable $getProjectDB): void - { - $dbForProject = $getProjectDB($project); - - try { - switch (true) { - case $document->getCollection() === 'users': // users - $sessions = count($document->getAttribute(METRIC_SESSIONS, 0)); - if (!empty($sessions)) { - $metrics[] = [ - 'key' => METRIC_SESSIONS, - 'value' => ($sessions * -1), - ]; - } - break; - case $document->getCollection() === 'databases': // databases - $collections = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{databaseInternalId}', $document->getInternalId(), METRIC_DATABASE_ID_COLLECTIONS))); - $documents = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{databaseInternalId}', $document->getInternalId(), METRIC_DATABASE_ID_DOCUMENTS))); - if (!empty($collections['value'])) { - $metrics[] = [ - 'key' => METRIC_COLLECTIONS, - 'value' => ($collections['value'] * -1), - ]; - } - - if (!empty($documents['value'])) { - $metrics[] = [ - 'key' => METRIC_DOCUMENTS, - 'value' => ($documents['value'] * -1), - ]; - } - break; - case str_starts_with($document->getCollection(), 'database_') && !str_contains($document->getCollection(), 'collection'): //collections - $parts = explode('_', $document->getCollection()); - $databaseInternalId = $parts[1] ?? 0; - $documents = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$databaseInternalId, $document->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_DOCUMENTS))); - - if (!empty($documents['value'])) { - $metrics[] = [ - 'key' => METRIC_DOCUMENTS, - 'value' => ($documents['value'] * -1), - ]; - $metrics[] = [ - 'key' => str_replace('{databaseInternalId}', $databaseInternalId, METRIC_DATABASE_ID_DOCUMENTS), - 'value' => ($documents['value'] * -1), - ]; - } - break; - - case $document->getCollection() === 'buckets': - $files = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{bucketInternalId}', $document->getInternalId(), METRIC_BUCKET_ID_FILES))); - $storage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{bucketInternalId}', $document->getInternalId(), METRIC_BUCKET_ID_FILES_STORAGE))); - - if (!empty($files['value'])) { - $metrics[] = [ - 'key' => METRIC_FILES, - 'value' => ($files['value'] * -1), - ]; - } - - if (!empty($storage['value'])) { - $metrics[] = [ - 'key' => METRIC_FILES_STORAGE, - 'value' => ($storage['value'] * -1), - ]; - } - break; - - case $document->getCollection() === 'functions': - $deployments = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{resourceType}', '{resourceInternalId}'], ['functions', $document->getInternalId()], METRIC_FUNCTION_ID_DEPLOYMENTS))); - $deploymentsStorage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{resourceType}', '{resourceInternalId}'], ['functions', $document->getInternalId()], METRIC_FUNCTION_ID_DEPLOYMENTS_STORAGE))); - $builds = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS))); - $buildsSuccess = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_SUCCESS))); - $buildsFailed = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_FAILED))); - $buildsStorage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_STORAGE))); - $buildsCompute = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE))); - $buildsComputeSuccess = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE_SUCCESS))); - $buildsComputeFailed = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE_FAILED))); - $executions = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_EXECUTIONS))); - $executionsCompute = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_EXECUTIONS_COMPUTE))); - - if (!empty($deployments['value'])) { - $metrics[] = [ - 'key' => METRIC_DEPLOYMENTS, - 'value' => ($deployments['value'] * -1), - ]; - } - - if (!empty($deploymentsStorage['value'])) { - $metrics[] = [ - 'key' => METRIC_DEPLOYMENTS_STORAGE, - 'value' => ($deploymentsStorage['value'] * -1), - ]; - } - - if (!empty($builds['value'])) { - $metrics[] = [ - 'key' => METRIC_BUILDS, - 'value' => ($builds['value'] * -1), - ]; - } - - if (!empty($buildsSuccess['value'])) { - $metrics[] = [ - 'key' => METRIC_BUILDS_SUCCESS, - 'value' => ($buildsSuccess['value'] * -1), - ]; - } - - if (!empty($buildsFailed['value'])) { - $metrics[] = [ - 'key' => METRIC_BUILDS_FAILED, - 'value' => ($buildsFailed['value'] * -1), - ]; - } - - if (!empty($buildsStorage['value'])) { - $metrics[] = [ - 'key' => METRIC_BUILDS_STORAGE, - 'value' => ($buildsStorage['value'] * -1), - ]; - } - - if (!empty($buildsCompute['value'])) { - $metrics[] = [ - 'key' => METRIC_BUILDS_COMPUTE, - 'value' => ($buildsCompute['value'] * -1), - ]; - } - - if (!empty($buildsComputeSuccess['value'])) { - $metrics[] = [ - 'key' => METRIC_BUILDS_COMPUTE_SUCCESS, - 'value' => ($buildsComputeSuccess['value'] * -1), - ]; - } - - if (!empty($buildsComputeFailed['value'])) { - $metrics[] = [ - 'key' => METRIC_BUILDS_COMPUTE_FAILED, - 'value' => ($buildsComputeFailed['value'] * -1), - ]; - } - - if (!empty($executions['value'])) { - $metrics[] = [ - 'key' => METRIC_EXECUTIONS, - 'value' => ($executions['value'] * -1), - ]; - } - - if (!empty($executionsCompute['value'])) { - $metrics[] = [ - 'key' => METRIC_EXECUTIONS_COMPUTE, - 'value' => ($executionsCompute['value'] * -1), - ]; - } - break; - default: - break; - } - } catch (\Throwable $e) { - console::error("[reducer] " . " {DateTime::now()} " . " {$project->getInternalId()} " . " {$e->getMessage()}"); - } - } -} diff --git a/src/Appwrite/Platform/Workers/UsageDump.php b/src/Appwrite/Platform/Workers/UsageDump.php deleted file mode 100644 index 2f1d13f29a..0000000000 --- a/src/Appwrite/Platform/Workers/UsageDump.php +++ /dev/null @@ -1,286 +0,0 @@ - 'Y-m-d H:00', - '1d' => 'Y-m-d 00:00', - 'inf' => '0000-00-00 00:00' - ]; - - public static function getName(): string - { - return 'usage-dump'; - } - - /** - * @throws \Exception - */ - public function __construct() - { - $this - ->inject('message') - ->inject('getProjectDB') - ->callback(function (Message $message, callable $getProjectDB) { - $this->action($message, $getProjectDB); - }); - } - - /** - * @param Message $message - * @param callable $getProjectDB - * @return void - * @throws Exception - * @throws \Utopia\Database\Exception - */ - public function action(Message $message, callable $getProjectDB): void - { - $payload = $message->getPayload() ?? []; - if (empty($payload)) { - throw new Exception('Missing payload'); - } - - - foreach ($payload['stats'] ?? [] as $stats) { - $project = new Document($stats['project'] ?? []); - - /** - * End temp bug fallback - */ - $numberOfKeys = !empty($stats['keys']) ? count($stats['keys']) : 0; - $receivedAt = $stats['receivedAt'] ?? 'NONE'; - if ($numberOfKeys === 0) { - continue; - } - - console::log('['.DateTime::now().'] Id: '.$project->getId(). ' InternalId: '.$project->getInternalId(). ' Db: '.$project->getAttribute('database').' ReceivedAt: '.$receivedAt. ' Keys: '.$numberOfKeys); - - try { - $dbForProject = $getProjectDB($project); - foreach ($stats['keys'] ?? [] as $key => $value) { - if ($value == 0) { - continue; - } - - if (str_contains($key, METRIC_DATABASES_STORAGE)) { - try { - $this->handleDatabaseStorage($key, $dbForProject); - } catch (\Exception $e) { - console::error('[' . DateTime::now() . '] failed to calculate database storage for key [' . $key . '] ' . $e->getMessage()); - } - continue; - } - - foreach ($this->periods as $period => $format) { - $time = 'inf' === $period ? null : date($format, time()); - $id = \md5("{$time}_{$period}_{$key}"); - - try { - $dbForProject->createDocument('stats', new Document([ - '$id' => $id, - 'period' => $period, - 'time' => $time, - 'metric' => $key, - 'value' => $value, - 'region' => System::getEnv('_APP_REGION', 'default'), - ])); - } catch (Duplicate $th) { - if ($value < 0) { - $dbForProject->decreaseDocumentAttribute( - 'stats', - $id, - 'value', - abs($value) - ); - } else { - $dbForProject->increaseDocumentAttribute( - 'stats', - $id, - 'value', - $value - ); - } - } - } - } - } catch (\Exception $e) { - console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage()); - } - } - } - - private function handleDatabaseStorage(string $key, Database $dbForProject): void - { - $data = explode('.', $key); - $start = microtime(true); - - $updateMetric = function (Database $dbForProject, int $value, string $key, string $period, string|null $time) { - $id = \md5("{$time}_{$period}_{$key}"); - - try { - $dbForProject->createDocument('stats', new Document([ - '$id' => $id, - 'period' => $period, - 'time' => $time, - 'metric' => $key, - 'value' => $value, - 'region' => System::getEnv('_APP_REGION', 'default'), - ])); - } catch (Duplicate $th) { - if ($value < 0) { - $dbForProject->decreaseDocumentAttribute( - 'stats', - $id, - 'value', - abs($value) - ); - } else { - $dbForProject->increaseDocumentAttribute( - 'stats', - $id, - 'value', - $value - ); - } - } - }; - - foreach ($this->periods as $period => $format) { - $time = 'inf' === $period ? null : date($format, time()); - $id = \md5("{$time}_{$period}_{$key}"); - - $value = 0; - $previousValue = 0; - try { - $previousValue = ($dbForProject->getDocument('stats', $id))->getAttribute('value', 0); - } catch (\Exception $e) { - // No previous value - } - - switch (count($data)) { - // Collection Level - case METRIC_COLLECTION_LEVEL_STORAGE: - Console::log('[' . DateTime::now() . '] Collection Level Storage Calculation [' . $key . ']'); - $databaseInternalId = $data[0]; - $collectionInternalId = $data[1]; - - try { - $value = $dbForProject->getSizeOfCollection('database_' . $databaseInternalId . '_collection_' . $collectionInternalId); - } catch (\Exception $e) { - // Collection not found - if ($e->getMessage() !== 'Collection not found') { - throw $e; - } - } - - // Compare with previous value - $diff = $value - $previousValue; - - if ($diff === 0) { - break; - } - - // Update Collection - $updateMetric($dbForProject, $diff, $key, $period, $time); - - // Update Database - $databaseKey = str_replace(['{databaseInternalId}'], [$data[0]], METRIC_DATABASE_ID_STORAGE); - $updateMetric($dbForProject, $diff, $databaseKey, $period, $time); - - // Update Project - $projectKey = METRIC_DATABASES_STORAGE; - $updateMetric($dbForProject, $diff, $projectKey, $period, $time); - break; - // Database Level - case METRIC_DATABASE_LEVEL_STORAGE: - Console::log('[' . DateTime::now() . '] Database Level Storage Calculation [' . $key . ']'); - $databaseInternalId = $data[0]; - - $collections = []; - try { - $collections = $dbForProject->find('database_' . $databaseInternalId); - } catch (\Exception $e) { - // Database not found - if ($e->getMessage() !== 'Collection not found') { - throw $e; - } - } - - foreach ($collections as $collection) { - try { - $value += $dbForProject->getSizeOfCollection('database_' . $databaseInternalId . '_collection_' . $collection->getInternalId()); - } catch (\Exception $e) { - // Collection not found - if ($e->getMessage() !== 'Collection not found') { - throw $e; - } - } - } - - $diff = $value - $previousValue; - - if ($diff === 0) { - break; - } - - // Update Database - $databaseKey = str_replace(['{databaseInternalId}'], [$data[0]], METRIC_DATABASE_ID_STORAGE); - $updateMetric($dbForProject, $diff, $databaseKey, $period, $time); - - // Update Project - $projectKey = METRIC_DATABASES_STORAGE; - $updateMetric($dbForProject, $diff, $projectKey, $period, $time); - break; - // Project Level - case METRIC_PROJECT_LEVEL_STORAGE: - Console::log('[' . DateTime::now() . '] Project Level Storage Calculation [' . $key . ']'); - // Get all project databases - $databases = $dbForProject->find('database'); - - // Recalculate all databases - foreach ($databases as $database) { - $collections = $dbForProject->find('database_' . $database->getInternalId()); - - foreach ($collections as $collection) { - try { - $value += $dbForProject->getSizeOfCollection('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId()); - } catch (\Exception $e) { - // Collection not found - if ($e->getMessage() !== 'Collection not found') { - throw $e; - } - } - } - } - - $diff = $value - $previousValue; - - // Update Project - $projectKey = METRIC_DATABASES_STORAGE; - $updateMetric($dbForProject, $diff, $projectKey, $period, $time); - break; - } - } - - $end = microtime(true); - - console::log('[' . DateTime::now() . '] DB Storage Calculation [' . $key . '] took ' . (($end - $start) * 1000) . ' milliseconds'); - } -}