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

fix: revert Traefik image version to 2.11; implement caching for function events and webhooks; add cache purging on function create/update/delete events

Browse Source
This commit is contained in:
shimon
2026-01-06 18:28:37 +02:00
parent cd651dbdb8
commit 23dfb23a3b
7 changed files with 183 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/controllers/shared/api.php
+33 -2
View File
@@ -112,6 +112,33 @@ $eventDatabaseListener = function (Document $project, Document $document, Respon
}
};
/**
* Purge function events cache when functions are created, updated or deleted.
*/
$functionsEventsCacheListener = function (string $event, Document $document, Document $project, Database $dbForProject) {
if ($document->getCollection() !== 'functions') {
return;
}
if ($project->isEmpty() || $project->getId() === 'console') {
return;
}
$hostname = $dbForProject->getAdapter()->getHostname();
$cacheKey = \sprintf(
'%s-cache-%s:%s:%s:project:%s:functions:events',
$dbForProject->getCacheName(),
$hostname ?? '',
$dbForProject->getNamespace(),
$dbForProject->getTenant(),
$project->getId()
);
var_dump(['purged' => $cacheKey]);
$dbForProject->getCache()->purge($cacheKey);
};
$usageDatabaseListener = function (string $event, Document $document, StatsUsage $queueForStatsUsage) {
$value = 1;
@@ -509,7 +536,7 @@ App::init()
->inject('devKey')
->inject('telemetry')
->inject('platform')
->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Publisher $publisher, Publisher $publisherFunctions, Publisher $publisherWebhooks, Event $queueForEvents, Messaging $queueForMessaging, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, StatsUsage $queueForStatsUsage, Func $queueForFunctions, Mail $queueForMails, Migration $queueForMigrations, Database $dbForProject, callable $timelimit, Document $resourceToken, string $mode, ?Key $apiKey, array $plan, Document $devKey, Telemetry $telemetry, array $platform) use ($usageDatabaseListener, $eventDatabaseListener) {
->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Publisher $publisher, Publisher $publisherFunctions, Publisher $publisherWebhooks, Event $queueForEvents, Messaging $queueForMessaging, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, StatsUsage $queueForStatsUsage, Func $queueForFunctions, Mail $queueForMails, Migration $queueForMigrations, Database $dbForProject, callable $timelimit, Document $resourceToken, string $mode, ?Key $apiKey, array $plan, Document $devKey, Telemetry $telemetry, array $platform) use ($usageDatabaseListener, $eventDatabaseListener, $functionsEventsCacheListener) {
$route = $utopia->getRoute();
@@ -650,7 +677,11 @@ App::init()
$queueForFunctions->from($queueForEvents),
$queueForWebhooks->from($queueForEvents),
$queueForRealtime->from($queueForEvents)
));
))
->on(Database::EVENT_DOCUMENT_CREATE, 'purge-function-events-cache', fn ($event, $document) => $functionsEventsCacheListener($event, $document, $project, $dbForProject))
->on(Database::EVENT_DOCUMENT_UPDATE, 'purge-function-events-cache', fn ($event, $document) => $functionsEventsCacheListener($event, $document, $project, $dbForProject))
->on(Database::EVENT_DOCUMENT_DELETE, 'purge-function-events-cache', fn ($event, $document) => $functionsEventsCacheListener($event, $document, $project, $dbForProject))
;
$useCache = $route->getLabel('cache', false);
$storageCacheOperationsCounter = $telemetry->createCounter('storage.cache.operations.load');
docker-compose.yml
+1 -2
View File
@@ -12,8 +12,7 @@ x-logging: &x-logging
services:
traefik:
#image: traefik:2.11 not working with docker api version 1.52
image: traefik:3.6
image: traefik:2.11
<<: *x-logging
container_name: appwrite-traefik
command:
src/Appwrite/Platform/Modules/Databases/Http/Databases/Action.php
+98
View File
@@ -7,6 +7,7 @@ use Appwrite\Platform\Action as AppwriteAction;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Operator;
use Utopia\Database\Query;
class Action extends AppwriteAction
{
@@ -94,4 +95,101 @@ class Action extends AppwriteAction
return $data;
}
/**
* Get function events for a project, using Redis cache
* @param Document|null $project
* @param Database $dbForProject
* @return array<string, bool>
*/
protected function getFunctionsEvents(?Document $project, Database $dbForProject): array
{
if ($project === null ||
$project->isEmpty() ||
$project->getId() === 'console') {
return [];
}
$hostname = $dbForProject->getAdapter()->getHostname();
$cacheKey = \sprintf(
'%s-cache-%s:%s:%s:project:%s:functions:events',
$dbForProject->getCacheName(),
$hostname ?? '',
$dbForProject->getNamespace(),
$dbForProject->getTenant(),
$project->getId()
);
$ttl = 3600; // 1 hour cache TTL
$cachedFunctionEvents = $dbForProject->getCache()->load($cacheKey, $ttl);
if ($cachedFunctionEvents !== false) {
return \json_decode($cachedFunctionEvents, true) ?? [];
}
try {
$events = [];
$limit = 100;
$sum = 100;
$offset = 0;
while ($sum >= $limit) {
$functions = $dbForProject->find('functions', [
Query::select(['$id', 'events']),
Query::limit($limit),
Query::offset($offset),
Query::orderAsc('$sequence'),
]);
$sum = \count($functions);
$offset = $offset + $limit;
foreach ($functions as $function) {
$functionEvents = $function->getAttribute('events', []);
if (!empty($functionEvents)) {
$events = array_merge($events, $functionEvents);
}
}
}
$uniqueEvents = \array_flip(\array_unique($events));
$dbForProject->getCache()->save($cacheKey, \json_encode($uniqueEvents));
return $uniqueEvents;
} catch (\Throwable $e) {
return [];
}
}
/**
* Get webhook events for a project from the project's webhooks attribute
* @param Document|null $project
* @return array<string, bool>
*/
protected function getWebhooksEvents(?Document $project): array
{
if ($project === null || $project->isEmpty() || $project->getId() === 'console') {
return [];
}
$webhooks = $project->getAttribute('webhooks', []);
if (empty($webhooks)) {
return [];
}
$events = [];
foreach ($webhooks as $webhook) {
if ($webhook->getAttribute('enabled', false) !== true) {
continue;
}
$webhookEvents = $webhook->getAttribute('events', []);
if (!empty($webhookEvents)) {
$events = array_merge($events, $webhookEvents);
}
}
return \array_flip(\array_unique($events));
}
}
src/Appwrite/Platform/Modules/Databases/Http/Databases/Collections/Documents/Action.php
+20 -98
View File
@@ -7,7 +7,6 @@ use Appwrite\Extend\Exception;
use Appwrite\Platform\Modules\Databases\Http\Databases\Action as DatabasesAction;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Query;
use Utopia\Database\Validator\Authorization;
abstract class Action extends DatabasesAction
@@ -373,8 +372,8 @@ abstract class Action extends DatabasesAction
// Get project and function events (cached)
$project = $queueForEvents->getProject();
$functionEvents = $this->getFunctionEvents($project, $dbForProject);
$webhookEvents = $this->getWebhookEvents($project);
$functionsEvents = $this->getFunctionsEvents($project, $dbForProject);
$webhooksEvents = $this->getWebhooksEvents($project);
foreach ($documents as $document) {
$queueForEvents
@@ -392,19 +391,26 @@ abstract class Action extends DatabasesAction
$queueForEvents->getParams()
);
// Only trigger functions if there are matching function events
if (!empty($functionEvents) && !empty(array_intersect($functionEvents, $generatedEvents))) {
$queueForFunctions
->from($queueForEvents)
->trigger();
if (!empty($functionsEvents)) {
foreach ($generatedEvents as $event) {
if (isset($functionsEvents[$event])) {
$queueForFunctions
->from($queueForEvents)
->trigger();
break;
}
}
}
// Only trigger webhooks if there are matching webhook events
if (!empty($webhookEvents) && !empty(array_intersect($webhookEvents, $generatedEvents))) {
$queueForWebhooks
->from($queueForEvents)
->trigger();
if (!empty($webhooksEvents)) {
foreach ($generatedEvents as $event) {
if (isset($webhooksEvents[$event])) {
$queueForWebhooks
->from($queueForEvents)
->trigger();
break;
}
}
}
}
@@ -413,88 +419,4 @@ abstract class Action extends DatabasesAction
$queueForFunctions->reset();
$queueForWebhooks->reset();
}
/**
* Get function events for a project, using Redis cache
* @param Document|null $project
* @param Database $dbForProject
* @return array
*/
protected function getFunctionEvents(?Document $project, Database $dbForProject): array
{
if ($project === null || $project->isEmpty() || $project->getId() === 'console') {
return [];
}
$hostname = $dbForProject->getAdapter()->getHostname();
$cacheKey = \sprintf(
'%s-cache-%s:%s:%s:project:%s:functionEvents',
$dbForProject->getCacheName(),
$hostname ?? '',
$dbForProject->getNamespace(),
$dbForProject->getTenant(),
$project->getId()
);
$ttl = 3600; // 1 hour cache TTL
$cachedFunctionEvents = $dbForProject->getCache()->load($cacheKey, $ttl);
if ($cachedFunctionEvents !== false) {
return \json_decode($cachedFunctionEvents, true) ?? [];
}
try {
$functions = $dbForProject->skipValidation(fn () => $dbForProject->find('functions', [
Query::limit(APP_LIMIT_SUBQUERY),
]));
$events = [];
foreach ($functions as $function) {
$functionEvents = $function->getAttribute('events', []);
if (!empty($functionEvents)) {
$events = array_merge($events, $functionEvents);
}
}
$uniqueEvents = array_unique($events);
// Save to cache
$dbForProject->getCache()->save($cacheKey, \json_encode($uniqueEvents), $ttl);
return $uniqueEvents;
} catch (\Throwable $e) {
return [];
}
}
/**
* Get webhook events for a project from the project's webhooks attribute
* @param Document|null $project
* @return array
*/
protected function getWebhookEvents(?Document $project): array
{
if ($project === null || $project->isEmpty() || $project->getId() === 'console') {
return [];
}
$webhooks = $project->getAttribute('webhooks', []);
if (empty($webhooks)) {
return [];
}
$events = [];
foreach ($webhooks as $webhook) {
if ($webhook->getAttribute('enabled', false) !== true) {
continue;
}
$webhookEvents = $webhook->getAttribute('events', []);
if (!empty($webhookEvents)) {
$events = array_merge($events, $webhookEvents);
}
}
return array_unique($events);
}
}
src/Appwrite/Platform/Modules/Databases/Http/Databases/Transactions/Update.php
+31 -2
View File
@@ -370,6 +370,11 @@ class Update extends Action
$queueForEvents->setEvent($eventString);
// Get project and function/webhook events (cached)
$project = $queueForEvents->getProject();
$functionsEvents = $this->getFunctionsEvents($project, $dbForProject);
$webhooksEvents = $this->getWebhooksEvents($project);
foreach ($documentsToTrigger as $doc) {
$payload = $doc->getArrayCopy();
$payload['$tableId'] = $collection->getId();
@@ -380,9 +385,33 @@ class Update extends Action
->setParam('rowId', $doc->getId())
->setPayload($payload);
// Generate events for this document operation
$generatedEvents = Event::generateEvents(
$queueForEvents->getEvent(),
$queueForEvents->getParams()
);
$queueForRealtime->from($queueForEvents)->trigger();
$queueForFunctions->from($queueForEvents)->trigger();
$queueForWebhooks->from($queueForEvents)->trigger();
// Only trigger functions if there are matching function events
if (!empty($functionsEvents)) {
foreach ($generatedEvents as $event) {
if (isset($functionsEvents[$event])) {
$queueForFunctions->from($queueForEvents)->trigger();
break;
}
}
}
// Only trigger webhooks if there are matching webhook events
if (!empty($webhooksEvents)) {
foreach ($generatedEvents as $event) {
if (isset($webhooksEvents[$event])) {
$queueForWebhooks->from($queueForEvents)->trigger();
break;
}
}
}
}
$queueForEvents->reset();
src/Appwrite/Platform/Modules/Functions/Http/Functions/Delete.php
-6
View File
@@ -13,7 +13,6 @@ use Appwrite\SDK\Response as SDKResponse;
use Appwrite\Utopia\Response;
use Utopia\Database\Database;
use Utopia\Database\DateTime;
use Utopia\Database\Document;
use Utopia\Database\Validator\Authorization;
use Utopia\Database\Validator\UID;
use Utopia\Platform\Action;
@@ -59,7 +58,6 @@ class Delete extends Base
->param('functionId', '', new UID(), 'Function ID.')
->inject('response')
->inject('dbForProject')
->inject('project')
->inject('queueForDeletes')
->inject('queueForEvents')
->inject('dbForPlatform')
@@ -70,7 +68,6 @@ class Delete extends Base
string $functionId,
Response $response,
Database $dbForProject,
Document $project,
DeleteEvent $queueForDeletes,
Event $queueForEvents,
Database $dbForPlatform
@@ -98,9 +95,6 @@ class Delete extends Base
$queueForEvents->setParam('functionId', $function->getId());
// Purge function events cache when function is deleted
$this->purgeFunctionEventsCache($project, $dbForProject);
$response->noContent();
}
}
src/Appwrite/Platform/Modules/Functions/Http/Functions/Update.php
-2
View File
@@ -286,8 +286,6 @@ class Update extends Base
$queueForEvents->setParam('functionId', $function->getId());
// Purge function events cache when function is updated
$this->purgeFunctionEventsCache($project, $dbForProject);
$response->dynamic($function, Response::MODEL_FUNCTION);
}