Appwrite/appwrite
1
0
Fork 0
mirror of https://github.com/appwrite/appwrite.git synced 2026-05-26 13:51:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: optimise events payloads

Browse Source
This commit is contained in:
Christy Jacob
2025-01-16 11:35:22 +05:30
parent 2f83bec071
commit de6fced412
11 changed files with 50 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Appwrite/Event/Event.php
+4 -1
View File
@@ -136,7 +136,10 @@ class Event
*/
public function setProject(Document $project): self
{
$this->project = $project;
$this->project = new Document([
'$id' => $project->getId(),
'$internalId' => $project->getInternalId()
]);
return $this;
}
src/Appwrite/Event/Webhook.php
+1 -5
View File
@@ -20,11 +20,7 @@ class Webhook extends Event
{
/** Filter out context and trim project to keep the payload small */
$this->context = [];
$this->project = new Document([
'$id' => $this->project->getId(),
'$internalId' => $this->project->getInternalId(),
]);
return parent::trigger();
}
}
src/Appwrite/Platform/Workers/Builds.php
+4 -3
View File
@@ -46,6 +46,7 @@ class Builds extends Action
$this
->desc('Builds worker')
->inject('message')
->inject('project')
->inject('dbForPlatform')
->inject('queueForEvents')
->inject('queueForFunctions')
@@ -54,11 +55,12 @@ class Builds extends Action
->inject('dbForProject')
->inject('deviceForFunctions')
->inject('log')
->callback(fn ($message, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, Usage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log) => $this->action($message, $dbForPlatform, $queueForEvents, $queueForFunctions, $usage, $cache, $dbForProject, $deviceForFunctions, $log));
->callback(fn ($message, Document $project, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, Usage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log) => $this->action($message, $project, $dbForPlatform, $queueForEvents, $queueForFunctions, $usage, $cache, $dbForProject, $deviceForFunctions, $log));
}
/**
* @param Message $message
* @param Document $project
* @param Database $dbForPlatform
* @param Event $queueForEvents
* @param Func $queueForFunctions
@@ -70,7 +72,7 @@ class Builds extends Action
* @return void
* @throws \Utopia\Database\Exception
*/
public function action(Message $message, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, Usage $queueForUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log): void
public function action(Message $message, Document $project, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, Usage $queueForUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log): void
{
$payload = $message->getPayload() ?? [];
@@ -79,7 +81,6 @@ class Builds extends Action
}
$type = $payload['type'] ?? '';
$project = new Document($payload['project'] ?? []);
$resource = new Document($payload['resource'] ?? []);
$deployment = new Document($payload['deployment'] ?? []);
$template = new Document($payload['template'] ?? []);
src/Appwrite/Platform/Workers/Databases.php
+4 -3
View File
@@ -34,21 +34,23 @@ class Databases extends Action
$this
->desc('Databases worker')
->inject('message')
->inject('project')
->inject('dbForPlatform')
->inject('dbForProject')
->inject('log')
->callback(fn (Message $message, Database $dbForPlatform, Database $dbForProject, Log $log) => $this->action($message, $dbForPlatform, $dbForProject, $log));
->callback(fn (Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Log $log) => $this->action($message, $project, $dbForPlatform, $dbForProject, $log));
}
/**
* @param Message $message
* @param Document $project
* @param Database $dbForPlatform
* @param Database $dbForProject
* @param Log $log
* @return void
* @throws \Exception
*/
public function action(Message $message, Database $dbForPlatform, Database $dbForProject, Log $log): void
public function action(Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Log $log): void
{
$payload = $message->getPayload() ?? [];
@@ -57,7 +59,6 @@ class Databases extends Action
}
$type = $payload['type'];
$project = new Document($payload['project']);
$collection = new Document($payload['collection'] ?? []);
$document = new Document($payload['document'] ?? []);
$database = new Document($payload['database'] ?? []);
src/Appwrite/Platform/Workers/Deletes.php
+4 -4
View File
@@ -43,6 +43,7 @@ class Deletes extends Action
$this
->desc('Deletes worker')
->inject('message')
->inject('project')
->inject('dbForPlatform')
->inject('getProjectDB')
->inject('timelimit')
@@ -55,8 +56,8 @@ class Deletes extends Action
->inject('auditRetention')
->inject('log')
->callback(
fn ($message, $dbForPlatform, callable $getProjectDB, callable $timelimit, Device $deviceForFiles, Device $deviceForFunctions, Device $deviceForBuilds, Device $deviceForCache, CertificatesAdapter $certificates, string $executionRetention, string $auditRetention, Log $log) =>
$this->action($message, $dbForPlatform, $getProjectDB, $timelimit, $deviceForFiles, $deviceForFunctions, $deviceForBuilds, $deviceForCache, $certificates, $executionRetention, $auditRetention, $log)
fn ($message, Document $project, Database $dbForPlatform, callable $getProjectDB, callable $timelimit, Device $deviceForFiles, Device $deviceForFunctions, Device $deviceForBuilds, Device $deviceForCache, CertificatesAdapter $certificates, string $executionRetention, string $auditRetention, Log $log) =>
$this->action($message, $project, $dbForPlatform, $getProjectDB, $timelimit, $deviceForFiles, $deviceForFunctions, $deviceForBuilds, $deviceForCache, $certificates, $executionRetention, $auditRetention, $log)
);
}
@@ -64,7 +65,7 @@ class Deletes extends Action
* @throws Exception
* @throws Throwable
*/
public function action(Message $message, Database $dbForPlatform, callable $getProjectDB, callable $timelimit, Device $deviceForFiles, Device $deviceForFunctions, Device $deviceForBuilds, Device $deviceForCache, CertificatesAdapter $certificates, string $executionRetention, string $auditRetention, Log $log): void
public function action(Message $message, Document $project, Database $dbForPlatform, callable $getProjectDB, callable $timelimit, Device $deviceForFiles, Device $deviceForFunctions, Device $deviceForBuilds, Device $deviceForCache, CertificatesAdapter $certificates, string $executionRetention, string $auditRetention, Log $log): void
{
$payload = $message->getPayload() ?? [];
@@ -78,7 +79,6 @@ class Deletes extends Action
$resource = $payload['resource'] ?? null;
$resourceType = $payload['resourceType'] ?? null;
$document = new Document($payload['document'] ?? []);
$project = new Document($payload['project'] ?? []);
$log->addTag('projectId', $project->getId());
$log->addTag('type', $type);
src/Appwrite/Platform/Workers/Messaging.php
+4 -2
View File
@@ -58,15 +58,17 @@ class Messaging extends Action
$this
->desc('Messaging worker')
->inject('message')
->inject('project')
->inject('log')
->inject('dbForProject')
->inject('deviceForFiles')
->inject('queueForUsage')
->callback(fn (Message $message, Log $log, Database $dbForProject, Device $deviceForFiles, Usage $queueForUsage) => $this->action($message, $log, $dbForProject, $deviceForFiles, $queueForUsage));
->callback(fn (Message $message, Document $project, Log $log, Database $dbForProject, Device $deviceForFiles, Usage $queueForUsage) => $this->action($message, $project, $log, $dbForProject, $deviceForFiles, $queueForUsage));
}
/**
* @param Message $message
* @param Document $project
* @param Log $log
* @param Database $dbForProject
* @param Device $deviceForFiles
@@ -76,6 +78,7 @@ class Messaging extends Action
*/
public function action(
Message $message,
Document $project,
Log $log,
Database $dbForProject,
Device $deviceForFiles,
@@ -89,7 +92,6 @@ class Messaging extends Action
}
$type = $payload['type'] ?? '';
$project = new Document($payload['project'] ?? []);
switch ($type) {
case MESSAGE_SEND_TYPE_INTERNAL:
src/Appwrite/Platform/Workers/Migrations.php
+3 -3
View File
@@ -51,16 +51,17 @@ class Migrations extends Action
$this
->desc('Migrations worker')
->inject('message')
->inject('project')
->inject('dbForProject')
->inject('dbForPlatform')
->inject('logError')
->callback(fn (Message $message, Database $dbForProject, Database $dbForPlatform, callable $logError) => $this->action($message, $dbForProject, $dbForPlatform, $logError));
->callback(fn (Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError) => $this->action($message, $project, $dbForProject, $dbForPlatform, $logError));
}
/**
* @throws Exception
*/
public function action(Message $message, Database $dbForProject, Database $dbForPlatform, callable $logError): void
public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError): void
{
$payload = $message->getPayload() ?? [];
@@ -69,7 +70,6 @@ class Migrations extends Action
}
$events = $payload['events'] ?? [];
$project = new Document($payload['project'] ?? []);
$migration = new Document($payload['migration'] ?? []);
if ($project->getId() === 'console') {
src/Appwrite/Platform/Workers/Usage.php
+5 -5
View File
@@ -34,10 +34,11 @@ class Usage extends Action
$this
->desc('Usage worker')
->inject('message')
->inject('project')
->inject('getProjectDB')
->inject('queueForUsageDump')
->callback(function (Message $message, callable $getProjectDB, UsageDump $queueForUsageDump) {
$this->action($message, $getProjectDB, $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');
@@ -46,21 +47,20 @@ class Usage extends Action
/**
* @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, callable $getProjectDB, UsageDump $queueForUsageDump): void
public function action(Message $message, Document $project, callable $getProjectDB, UsageDump $queueForUsageDump): void
{
$payload = $message->getPayload() ?? [];
if (empty($payload)) {
throw new Exception('Missing payload');
}
$document = $payload['project'] ?? [];
$project = new Document($document);
if (empty($project->getAttribute('database'))) {
var_dump($payload);
src/Appwrite/Platform/Workers/UsageDump.php
+1 -16
View File
@@ -59,26 +59,11 @@ class UsageDump extends Action
foreach ($payload['stats'] ?? [] as $stats) {
//$project = new Document($stats['project'] ?? []);
/**
* Start temp bug fallback
*/
$document = $stats['project'] ?? [];
if (!empty($document['$uid'])) {
$document['$id'] = $document['$uid'];
}
$project = new Document($document);
if (empty($project->getAttribute('database'))) {
continue;
}
$project = new Document($stats['project'] ?? []);
/**
* End temp bug fallback
*/
$numberOfKeys = !empty($stats['keys']) ? count($stats['keys']) : 0;
$receivedAt = $stats['receivedAt'] ?? 'NONE';
if ($numberOfKeys === 0) {
src/Appwrite/Platform/Workers/Webhooks.php
+4 -4
View File
@@ -32,22 +32,24 @@ class Webhooks extends Action
$this
->desc('Webhooks worker')
->inject('message')
->inject('project')
->inject('dbForPlatform')
->inject('queueForMails')
->inject('queueForUsage')
->inject('log')
->callback(fn (Message $message, Database $dbForPlatform, Mail $queueForMails, Usage $queueForUsage, Log $log) => $this->action($message, $dbForPlatform, $queueForMails, $queueForUsage, $log));
->callback(fn (Message $message, Document $project, Database $dbForPlatform, Mail $queueForMails, Usage $queueForUsage, Log $log) => $this->action($message, $project, $dbForPlatform, $queueForMails, $queueForUsage, $log));
}
/**
* @param Message $message
* @param Document $project
* @param Database $dbForPlatform
* @param Mail $queueForMails
* @param Log $log
* @return void
* @throws Exception
*/
public function action(Message $message, Database $dbForPlatform, Mail $queueForMails, Usage $queueForUsage, Log $log): void
public function action(Message $message, Document $project, Database $dbForPlatform, Mail $queueForMails, Usage $queueForUsage, Log $log): void
{
$this->errors = [];
$payload = $message->getPayload() ?? [];
@@ -60,8 +62,6 @@ class Webhooks extends Action
$webhookPayload = json_encode($payload['payload']);
$user = new Document($payload['user'] ?? []);
$project = new Document($payload['project']);
$project = $dbForPlatform->getDocument('projects', $project->getId());
$log->addTag('projectId', $project->getId());
foreach ($project->getAttribute('webhooks', []) as $webhook) {