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

Merge remote-tracking branch 'origin/1.6.x' into feat-logs-db

Browse Source
This commit is contained in:
Damodar Lohani
2025-02-05 09:49:11 +00:00
parent 966fc47e25 1adb1e23a9
commit cf66bee925
31 changed files with 551 additions and 491 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Appwrite/Event/Audit.php
+3 -3
View File
@@ -2,7 +2,7 @@
namespace Appwrite\Event;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Audit extends Event
{
@@ -12,9 +12,9 @@ class Audit extends Event
protected string $ip = '';
protected string $hostname = '';
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::AUDITS_QUEUE_NAME)
src/Appwrite/Event/Build.php
+3 -3
View File
@@ -3,7 +3,7 @@
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Build extends Event
{
@@ -12,9 +12,9 @@ class Build extends Event
protected ?Document $deployment = null;
protected ?Document $template = null;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::BUILDS_QUEUE_NAME)
src/Appwrite/Event/Certificate.php
+3 -3
View File
@@ -3,16 +3,16 @@
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Certificate extends Event
{
protected bool $skipRenewCheck = false;
protected ?Document $domain = null;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::CERTIFICATES_QUEUE_NAME)
src/Appwrite/Event/Database.php
+3 -3
View File
@@ -4,7 +4,7 @@ namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\DSN\DSN;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Database extends Event
{
@@ -13,9 +13,9 @@ class Database extends Event
protected ?Document $collection = null;
protected ?Document $document = null;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this->setClass(Event::DATABASE_CLASS_NAME);
}
src/Appwrite/Event/Delete.php
+3 -3
View File
@@ -3,7 +3,7 @@
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Delete extends Event
{
@@ -15,9 +15,9 @@ class Delete extends Event
protected ?string $hourlyUsageRetentionDatetime = null;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::DELETE_QUEUE_NAME)
src/Appwrite/Event/Event.php
+6 -7
View File
@@ -4,8 +4,8 @@ namespace Appwrite\Event;
use InvalidArgumentException;
use Utopia\Database\Document;
use Utopia\Queue\Client;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
use Utopia\Queue\Queue;
class Event
{
@@ -69,10 +69,10 @@ class Event
protected bool $paused = false;
/**
* @param Connection $connection
* @param Publisher $publisher
* @return void
*/
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
}
@@ -356,12 +356,11 @@ class Event
}
/** The getter is required since events like Databases need to override the queue name depending on the project */
$client = new Client($this->getQueue(), $this->connection);
$queue = new Queue($this->getQueue());
// Merge the base payload with any trimmed values
$payload = array_merge($this->preparePayload(), $this->trimPayload());
return $client->enqueue($payload);
return $this->publisher->enqueue($queue, $payload);
}
/**
src/Appwrite/Event/Func.php
+3 -3
View File
@@ -3,7 +3,7 @@
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Func extends Event
{
@@ -19,9 +19,9 @@ class Func extends Event
protected ?Document $function = null;
protected ?Document $execution = null;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::FUNCTIONS_QUEUE_NAME)
src/Appwrite/Event/Mail.php
+3 -3
View File
@@ -2,7 +2,7 @@
namespace Appwrite\Event;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Mail extends Event
{
@@ -15,9 +15,9 @@ class Mail extends Event
protected string $bodyTemplate = '';
protected array $attachment = [];
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::MAILS_QUEUE_NAME)
src/Appwrite/Event/Messaging.php
+3 -3
View File
@@ -3,7 +3,7 @@
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Messaging extends Event
{
@@ -14,9 +14,9 @@ class Messaging extends Event
protected ?string $scheduledAt = null;
protected ?string $providerType = null;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::MESSAGING_QUEUE_NAME)
src/Appwrite/Event/Migration.php
+3 -3
View File
@@ -3,16 +3,16 @@
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Migration extends Event
{
protected string $type = '';
protected ?Document $migration = null;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::MIGRATIONS_QUEUE_NAME)
src/Appwrite/Event/Usage.php
+3 -3
View File
@@ -3,16 +3,16 @@
namespace Appwrite\Event;
use Utopia\Database\Document;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Usage extends Event
{
protected array $metrics = [];
protected array $reduce = [];
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::USAGE_QUEUE_NAME)
src/Appwrite/Event/UsageDump.php
+3 -3
View File
@@ -2,15 +2,15 @@
namespace Appwrite\Event;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class UsageDump extends Event
{
protected array $stats;
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::USAGE_DUMP_QUEUE_NAME)
src/Appwrite/Event/Webhook.php
+3 -3
View File
@@ -2,13 +2,13 @@
namespace Appwrite\Event;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
class Webhook extends Event
{
public function __construct(protected Connection $connection)
public function __construct(protected Publisher $publisher)
{
parent::__construct($connection);
parent::__construct($publisher);
$this
->setQueue(Event::WEBHOOK_QUEUE_NAME)
src/Appwrite/Platform/Services/Tasks.php
-2
View File
@@ -6,7 +6,6 @@ use Appwrite\Platform\Tasks\Doctor;
use Appwrite\Platform\Tasks\Install;
use Appwrite\Platform\Tasks\Maintenance;
use Appwrite\Platform\Tasks\Migrate;
use Appwrite\Platform\Tasks\QueueCount;
use Appwrite\Platform\Tasks\QueueRetry;
use Appwrite\Platform\Tasks\ScheduleExecutions;
use Appwrite\Platform\Tasks\ScheduleFunctions;
@@ -30,7 +29,6 @@ class Tasks extends Service
->addAction(Install::getName(), new Install())
->addAction(Maintenance::getName(), new Maintenance())
->addAction(Migrate::getName(), new Migrate())
->addAction(QueueCount::getName(), new QueueCount())
->addAction(QueueRetry::getName(), new QueueRetry())
->addAction(SDKs::getName(), new SDKs())
->addAction(SSL::getName(), new SSL())
src/Appwrite/Platform/Tasks/QueueCount.php
-57
View File
@@ -1,57 +0,0 @@
<?php
namespace Appwrite\Platform\Tasks;
use Utopia\CLI\Console;
use Utopia\Platform\Action;
use Utopia\Queue\Client;
use Utopia\Queue\Connection;
use Utopia\Validator\Text;
use Utopia\Validator\WhiteList;
class QueueCount extends Action
{
public static function getName(): string
{
return 'queue-count';
}
public function __construct()
{
$this
->desc('Return the number of from a specific queue identified by the name parameter with a specific type')
->param('name', '', new Text(100), 'Queue name')
->param('type', '', new WhiteList([
'success',
'failed',
'processing',
]), 'Queue type')
->inject('queue')
->callback(fn ($name, $type, $queue) => $this->action($name, $type, $queue));
}
/**
* @param string $name The name of the queue to count the jobs from
* @param string $type The type of jobs to count
* @param Connection $queue
*/
public function action(string $name, string $type, Connection $queue): void
{
if (!$name) {
Console::error('Missing required parameter $name');
return;
}
$queueClient = new Client($name, $queue);
$count = match ($type) {
'success' => $queueClient->countSuccessfulJobs(),
'failed' => $queueClient->countFailedJobs(),
'processing' => $queueClient->countProcessingJobs(),
default => 0
};
Console::log("Queue: '{$name}' has {$count} {$type} jobs.");
}
}
src/Appwrite/Platform/Tasks/QueueRetry.php
+7 -16
View File
@@ -4,8 +4,8 @@ namespace Appwrite\Platform\Tasks;
use Utopia\CLI\Console;
use Utopia\Platform\Action;
use Utopia\Queue\Client;
use Utopia\Queue\Connection;
use Utopia\Queue\Publisher;
use Utopia\Queue\Queue;
use Utopia\Validator\Text;
use Utopia\Validator\Wildcard;
@@ -23,33 +23,24 @@ class QueueRetry extends Action
->desc('Retry failed jobs from a specific queue identified by the name parameter')
->param('name', '', new Text(100), 'Queue name')
->param('limit', 0, new Wildcard(), 'jobs limit', true)
->inject('queue')
->callback(fn ($name, $limit, $queue) => $this->action($name, $limit, $queue));
->inject('publisher')
->callback(fn ($name, $limit, $publisher) => $this->action($name, $limit, $publisher));
}
/**
* @param string $name The name of the queue to retry jobs from
* @param mixed $limit
* @param Connection $queue
* @param Publisher $publisher
*/
public function action(string $name, mixed $limit, Connection $queue): void
public function action(string $name, mixed $limit, Publisher $publisher): void
{
if (!$name) {
Console::error('Missing required parameter $name');
return;
}
$limit = (int)$limit;
$queueClient = new Client($name, $queue);
if ($queueClient->countFailedJobs() === 0) {
Console::error('No failed jobs found.');
return;
}
Console::log('Retrying failed jobs...');
$queueClient->retry($limit);
$publisher->retry(new Queue($name), $limit);
}
}
src/Appwrite/Platform/Tasks/ScheduleBase.php
+9 -13
View File
@@ -11,7 +11,7 @@ use Utopia\Database\Exception;
use Utopia\Database\Query;
use Utopia\Database\Validator\Authorization;
use Utopia\Platform\Action;
use Utopia\Pools\Group;
use Utopia\Queue\Publisher;
use Utopia\System\System;
use function Swoole\Coroutine\run;
@@ -26,7 +26,7 @@ abstract class ScheduleBase extends Action
abstract public static function getName(): string;
abstract public static function getSupportedResource(): string;
abstract public static function getCollectionId(): string;
abstract protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void;
abstract protected function enqueueResources(Publisher $publisher, Database $dbForPlatform, callable $getProjectDB): void;
public function __construct()
{
@@ -34,10 +34,10 @@ abstract class ScheduleBase extends Action
$this
->desc("Execute {$type}s scheduled in Appwrite")
->inject('pools')
->inject('publisher')
->inject('dbForPlatform')
->inject('getProjectDB')
->callback(fn (Group $pools, Database $dbForPlatform, callable $getProjectDB) => $this->action($pools, $dbForPlatform, $getProjectDB));
->callback(fn (Publisher $publisher, Database $dbForPlatform, callable $getProjectDB) => $this->action($publisher, $dbForPlatform, $getProjectDB));
}
protected function updateProjectAccess(Document $project, Database $dbForPlatform): void
@@ -56,7 +56,7 @@ abstract class ScheduleBase extends Action
* 2. Create timer that sync all changes from 'schedules' collection to local copy. Only reading changes thanks to 'resourceUpdatedAt' attribute
* 3. Create timer that prepares coroutines for soon-to-execute schedules. When it's ready, coroutine sleeps until exact time before sending request to worker.
*/
public function action(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
public function action(Publisher $publisher, Database $dbForPlatform, callable $getProjectDB): void
{
Console::title(\ucfirst(static::getSupportedResource()) . ' scheduler V1');
Console::success(APP_NAME . ' ' . \ucfirst(static::getSupportedResource()) . ' scheduler v1 has started');
@@ -125,17 +125,15 @@ abstract class ScheduleBase extends Action
$latestDocument = \end($results);
}
$pools->reclaim();
Console::success("{$total} resources were loaded in " . (\microtime(true) - $loadStart) . " seconds");
Console::success("Starting timers at " . DateTime::now());
run(function () use ($dbForPlatform, &$lastSyncUpdate, $getSchedule, $pools, $getProjectDB) {
run(function () use ($dbForPlatform, &$lastSyncUpdate, $getSchedule, $publisher, $getProjectDB) {
/**
* The timer synchronize $schedules copy with database collection.
*/
Timer::tick(static::UPDATE_TIMER * 1000, function () use ($dbForPlatform, &$lastSyncUpdate, $getSchedule, $pools) {
Timer::tick(static::UPDATE_TIMER * 1000, function () use ($dbForPlatform, &$lastSyncUpdate, $getSchedule) {
$time = DateTime::now();
$timerStart = \microtime(true);
@@ -184,17 +182,15 @@ abstract class ScheduleBase extends Action
$lastSyncUpdate = $time;
$timerEnd = \microtime(true);
$pools->reclaim();
Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds");
});
Timer::tick(
static::ENQUEUE_TIMER * 1000,
fn () => $this->enqueueResources($pools, $dbForPlatform, $getProjectDB)
fn () => $this->enqueueResources($publisher, $dbForPlatform, $getProjectDB)
);
$this->enqueueResources($pools, $dbForPlatform, $getProjectDB);
$this->enqueueResources($publisher, $dbForPlatform, $getProjectDB);
});
}
}
src/Appwrite/Platform/Tasks/ScheduleExecutions.php
+3 -7
View File
@@ -5,7 +5,7 @@ namespace Appwrite\Platform\Tasks;
use Appwrite\Event\Func;
use Swoole\Coroutine as Co;
use Utopia\Database\Database;
use Utopia\Pools\Group;
use Utopia\Queue\Publisher;
class ScheduleExecutions extends ScheduleBase
{
@@ -27,11 +27,9 @@ class ScheduleExecutions extends ScheduleBase
return 'executions';
}
protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
protected function enqueueResources(Publisher $publisher, Database $dbForPlatform, callable $getProjectDB): void
{
$queue = $pools->get('queue')->pop();
$connection = $queue->getResource();
$queueForFunctions = new Func($connection);
$queueForFunctions = new Func($publisher);
$intervalEnd = (new \DateTime())->modify('+' . self::ENQUEUE_TIMER . ' seconds');
foreach ($this->schedules as $schedule) {
@@ -83,7 +81,5 @@ class ScheduleExecutions extends ScheduleBase
unset($this->schedules[$schedule['$internalId']]);
}
$queue->reclaim();
}
}
src/Appwrite/Platform/Tasks/ScheduleFunctions.php
+4 -10
View File
@@ -7,7 +7,7 @@ use Cron\CronExpression;
use Utopia\CLI\Console;
use Utopia\Database\Database;
use Utopia\Database\DateTime;
use Utopia\Pools\Group;
use Utopia\Queue\Publisher;
class ScheduleFunctions extends ScheduleBase
{
@@ -31,7 +31,7 @@ class ScheduleFunctions extends ScheduleBase
return 'functions';
}
protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
protected function enqueueResources(Publisher $publisher, Database $dbForPlatform, callable $getProjectDB): void
{
$timerStart = \microtime(true);
$time = DateTime::now();
@@ -70,12 +70,9 @@ class ScheduleFunctions extends ScheduleBase
}
foreach ($delayedExecutions as $delay => $scheduleKeys) {
\go(function () use ($delay, $scheduleKeys, $pools, $dbForPlatform) {
\go(function () use ($delay, $scheduleKeys, $publisher, $dbForPlatform) {
\sleep($delay); // in seconds
$queue = $pools->get('queue')->pop();
$connection = $queue->getResource();
foreach ($scheduleKeys as $scheduleKey) {
// Ensure schedule was not deleted
if (!\array_key_exists($scheduleKey, $this->schedules)) {
@@ -86,8 +83,7 @@ class ScheduleFunctions extends ScheduleBase
$this->updateProjectAccess($schedule['project'], $dbForPlatform);
$queueForFunctions = new Func($connection);
$queueForFunctions = new Func($publisher);
$queueForFunctions
->setType('schedule')
->setFunction($schedule['resource'])
@@ -96,8 +92,6 @@ class ScheduleFunctions extends ScheduleBase
->setProject($schedule['project'])
->trigger();
}
$queue->reclaim();
});
}
src/Appwrite/Platform/Tasks/ScheduleMessages.php
+4 -10
View File
@@ -4,7 +4,7 @@ namespace Appwrite\Platform\Tasks;
use Appwrite\Event\Messaging;
use Utopia\Database\Database;
use Utopia\Pools\Group;
use Utopia\Queue\Publisher;
class ScheduleMessages extends ScheduleBase
{
@@ -26,7 +26,7 @@ class ScheduleMessages extends ScheduleBase
return 'messages';
}
protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
protected function enqueueResources(Publisher $publisher, Database $dbForPlatform, callable $getProjectDB): void
{
foreach ($this->schedules as $schedule) {
if (!$schedule['active']) {
@@ -40,13 +40,9 @@ class ScheduleMessages extends ScheduleBase
continue;
}
\go(function () use ($schedule, $pools, $dbForPlatform) {
$queue = $pools->get('queue')->pop();
$connection = $queue->getResource();
$queueForMessaging = new Messaging($connection);
\go(function () use ($schedule, $publisher, $dbForPlatform) {
$this->updateProjectAccess($schedule['project'], $dbForPlatform);
$queueForMessaging = new Messaging($publisher);
$queueForMessaging
->setType(MESSAGE_SEND_TYPE_EXTERNAL)
->setMessageId($schedule['resourceId'])
@@ -58,8 +54,6 @@ class ScheduleMessages extends ScheduleBase
$schedule['$id'],
);
$queue->reclaim();
unset($this->schedules[$schedule['$internalId']]);
});
}