Merge pull request #10156 from appwrite/feat-stats-usage-on-redis

feat: stats-usage on redis

CONTRIBUTING.md

@@ -9,7 +9,31 @@ You can [find issues using this query](https://github.com/search?q=org%3Aappwrit
 
 ## How to Start?
 
-If you are worried or don’t know where to start, check out the next section that explains what kind of help we could use and where you can get involved. You can send your questions to [@appwrite on Twitter](https://twitter.com/appwrite) or to anyone from the [Appwrite team on Discord](https://appwrite.io/discord). You can also submit an issue, and a maintainer can guide you!
+Welcome! We're excited that you're interested in contributing to Appwrite. To make sure your time is valued and your contributions are successful, please follow these steps before writing any code:
+
+### 🔍 Step 1: Find an Issue
+
+Browse open issues and look for ones labeled [good first issue](https://github.com/search?q=org%3Aappwrite+is%3Aopen+type%3Aissue+label%3A%22good+first+issue%22&type=issues) or [help wanted](https://github.com/search?q=org%3Aappwrite+is%3Aopen+type%3Aissue+label%3A%22help+wanted%22&type=issues).
+
+If you're not sure which issue to pick, ask in our ⁠[maintainers channel](https://discord.com/channels/564160730845151244/636852860709240842) on Discord.
+
+### 📝 Step 2: Ask to Be Assigned
+
+Before working on an issue, comment on the GitHub issue asking to be assigned. This prevents multiple people working on the same task.
+
+Then, create a thread in the ⁠[maintainers channel](https://discord.com/channels/564160730845151244/636852860709240842) on Discord with a link to the issue.
+
+Our team is small and may not see your GitHub comment right away - posting in the ⁠[maintainers channel](https://discord.com/channels/564160730845151244/636852860709240842) ensures it gets seen.
+
+### 💬 Step 3: Don’t Submit Random PRs
+
+If you're not working on an assigned issue, create a GitHub issue first.
+
+PRs submitted without context or discussion may not align with our roadmap and may be closed without review.
+
+### ⚠️ Please Note
+
+We’re a very small team managing a large project. Many PRs are submitted, and while we appreciate every effort, we can only review contributions that follow the process above. This helps us keep things fair and organized.
 
 ## Code of Conduct
 

app/cli.php

@@ -191,9 +191,15 @@ CLI::setResource('getLogsDB', function (Group $pools, Cache $cache) {
 CLI::setResource('publisher', function (Group $pools) {
     return new BrokerPool(publisher: $pools->get('publisher'));
 }, ['pools']);
-CLI::setResource('publisherRedis', function () {
-    // Stub
-});
+CLI::setResource('publisherDatabases', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
+CLI::setResource('publisherMigrations', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
+CLI::setResource('publisherStatsUsage', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
 CLI::setResource('queueForStatsUsage', function (Publisher $publisher) {
     return new StatsUsage($publisher);
 }, ['publisher']);

app/controllers/api/health.php

@@ -522,17 +522,11 @@ App::get('/v1/health/queue/databases')
     ))
     ->param('name', 'database_db_main', new Text(256), 'Queue name for which to check the queue size', true)
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('publisher')
-    ->inject('publisherRedis')
+    ->inject('publisherDatabases')
     ->inject('response')
-    ->action(function (string $name, int|string $threshold, Publisher $publisher, ?Publisher $publisherRedis, Response $response) {
+    ->action(function (string $name, int|string $threshold, Publisher $publisherDatabases, Response $response) {
         $threshold = \intval($threshold);
-
-        $isRedisFallback = \str_contains(System::getEnv('_APP_WORKER_REDIS_FALLBACK', ''), 'databases');
-
-        $size = $isRedisFallback
-            ? $publisherRedis->getQueueSize(new Queue($name))
-            : $publisher->getQueueSize(new Queue($name));
+        $size = $publisherDatabases->getQueueSize(new Queue($name));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -659,17 +653,12 @@ App::get('/v1/health/queue/migrations')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('publisher')
-    ->inject('publisherRedis')
+    ->inject('publisherMigrations')
     ->inject('response')
-    ->action(function (int|string $threshold, Publisher $publisher, ?Publisher $publisherRedis, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisherMigrations, Response $response) {
         $threshold = \intval($threshold);
 
-        $isRedisFallback = \str_contains(System::getEnv('_APP_WORKER_REDIS_FALLBACK', ''), 'migrations');
-
-        $size = $isRedisFallback
-            ? $publisherRedis->getQueueSize(new Queue(Event::MIGRATIONS_QUEUE_NAME))
-            : $publisher->getQueueSize(new Queue(Event::MIGRATIONS_QUEUE_NAME));
+        $size = $publisherMigrations->getQueueSize(new Queue(Event::MIGRATIONS_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");

app/init/resources.php

@@ -80,15 +80,27 @@ App::setResource('localeCodes', function () {
 App::setResource('publisher', function (Group $pools) {
     return new BrokerPool(publisher: $pools->get('publisher'));
 }, ['pools']);
-App::setResource('publisherRedis', function () {
-    // Stub
-});
+App::setResource('publisherDatabases', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
+App::setResource('publisherMigrations', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
+App::setResource('publisherStatsUsage', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
 App::setResource('consumer', function (Group $pools) {
     return new BrokerPool(consumer: $pools->get('consumer'));
 }, ['pools']);
-App::setResource('consumerRedis', function () {
-    // Stub
-});
+App::setResource('consumerDatabases', function (BrokerPool $consumer) {
+    return $consumer;
+}, ['consumer']);
+App::setResource('consumerMigrations', function (BrokerPool $consumer) {
+    return $consumer;
+}, ['publisher']);
+App::setResource('consumerStatsUsage', function (BrokerPool $consumer) {
+    return $consumer;
+}, ['publisher']);
 App::setResource('queueForMessaging', function (Publisher $publisher) {
     return new Messaging($publisher);
 }, ['publisher']);

app/worker.php

@@ -247,17 +247,33 @@ Server::setResource('publisher', function (Group $pools) {
     return new BrokerPool(publisher: $pools->get('publisher'));
 }, ['pools']);
 
-Server::setResource('publisherRedis', function () {
-    // Stub
-});
+Server::setResource('publisherDatabases', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
+
+Server::setResource('publisherMigrations', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
+
+Server::setResource('publisherStatsUsage', function (BrokerPool $publisher) {
+    return $publisher;
+}, ['publisher']);
 
 Server::setResource('consumer', function (Group $pools) {
     return new BrokerPool(consumer: $pools->get('consumer'));
 }, ['pools']);
 
-Server::setResource('consumerRedis', function () {
-    // Stub
-});
+Server::setResource('consumerDatabases', function (BrokerPool $consumer) {
+    return $consumer;
+}, ['consumer']);
+
+Server::setResource('consumerMigrations', function (BrokerPool $consumer) {
+    return $consumer;
+}, ['consumer']);
+
+Server::setResource('consumerStatsUsage', function (BrokerPool $consumer) {
+    return $consumer;
+}, ['consumer']);
 
 Server::setResource('queueForStatsUsage', function (Publisher $publisher) {
     return new StatsUsage($publisher);

src/Appwrite/Platform/Tasks/ScheduleBase.php

@@ -11,7 +11,6 @@ use Utopia\Database\Exception;
 use Utopia\Database\Query;
 use Utopia\Database\Validator\Authorization;
 use Utopia\Platform\Action;
-use Utopia\Pools\Group;
 use Utopia\Queue\Broker\Pool as BrokerPool;
 use Utopia\System\System;
 use Utopia\Telemetry\Adapter as Telemetry;
@@ -26,7 +25,7 @@ abstract class ScheduleBase extends Action
     protected array $schedules = [];
 
     protected BrokerPool $publisher;
-    protected ?BrokerPool $publisherRedis = null;
+    protected BrokerPool $publisherMigrations;
 
     private ?Histogram $collectSchedulesTelemetryDuration = null;
     private ?Gauge $collectSchedulesTelemetryCount = null;
@@ -36,7 +35,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(Database $dbForPlatform, callable $getProjectDB): void;
 
     public function __construct()
     {
@@ -44,7 +43,8 @@ abstract class ScheduleBase extends Action
 
         $this
             ->desc("Execute {$type}s scheduled in Appwrite")
-            ->inject('pools')
+            ->inject('publisher')
+            ->inject('publisherMigrations')
             ->inject('dbForPlatform')
             ->inject('getProjectDB')
             ->inject('telemetry')
@@ -67,18 +67,13 @@ 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, Telemetry $telemetry): void
+    public function action(BrokerPool $publisher, BrokerPool $publisherMigrations, Database $dbForPlatform, callable $getProjectDB, Telemetry $telemetry): void
     {
         Console::title(\ucfirst(static::getSupportedResource()) . ' scheduler V1');
         Console::success(APP_NAME . ' ' . \ucfirst(static::getSupportedResource()) . ' scheduler v1 has started');
 
-        $this->publisher = new BrokerPool($pools->get('publisher'));
-
-        try {
-            $this->publisherRedis = new BrokerPool($pools->get('publisherRedis'));
-        } catch (\Throwable) {
-            $this->publisherRedis = null;
-        }
+        $this->publisher = $publisher;
+        $this->publisherMigrations = $publisherMigrations;
 
         $this->scheduleTelemetryCount = $telemetry->createGauge('task.schedule.count');
         $this->collectSchedulesTelemetryDuration = $telemetry->createHistogram('task.schedule.collect_schedules.duration', 's');
@@ -101,7 +96,7 @@ abstract class ScheduleBase extends Action
 
         while (true) {
             try {
-                go(fn () => $this->enqueueResources($pools, $dbForPlatform, $getProjectDB));
+                go(fn () => $this->enqueueResources($dbForPlatform, $getProjectDB));
                 $this->scheduleTelemetryCount->record(count($this->schedules), ['resourceType' => static::getSupportedResource()]);
                 sleep(static::ENQUEUE_TIMER);
             } catch (\Throwable $th) {

src/Appwrite/Platform/Tasks/ScheduleExecutions.php

@@ -5,8 +5,6 @@ namespace Appwrite\Platform\Tasks;
 use Appwrite\Event\Func;
 use Swoole\Coroutine as Co;
 use Utopia\Database\Database;
-use Utopia\Pools\Group;
-use Utopia\System\System;
 
 class ScheduleExecutions extends ScheduleBase
 {
@@ -28,17 +26,11 @@ class ScheduleExecutions extends ScheduleBase
         return 'executions';
     }
 
-    protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
+    protected function enqueueResources(Database $dbForPlatform, callable $getProjectDB): void
     {
         $intervalEnd = (new \DateTime())->modify('+' . self::ENQUEUE_TIMER . ' seconds');
 
-        $isRedisFallback = \str_contains(System::getEnv('_APP_WORKER_REDIS_FALLBACK', ''), 'functions');
-
-        $queueForFunctions = new Func(
-            $isRedisFallback
-            ? $this->publisherRedis
-            : $this->publisher
-        );
+        $queueForFunctions = new Func($this->publisher);
 
         foreach ($this->schedules as $schedule) {
             if (!$schedule['active']) {

src/Appwrite/Platform/Tasks/ScheduleFunctions.php

@@ -8,7 +8,6 @@ use Utopia\CLI\Console;
 use Utopia\Database\Database;
 use Utopia\Database\DateTime;
 use Utopia\Pools\Group;
-use Utopia\System\System;
 
 class ScheduleFunctions extends ScheduleBase
 {
@@ -32,7 +31,7 @@ class ScheduleFunctions extends ScheduleBase
         return 'functions';
     }
 
-    protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
+    protected function enqueueResources(Database $dbForPlatform, callable $getProjectDB): void
     {
         $timerStart = \microtime(true);
         $time = DateTime::now();
@@ -91,13 +90,7 @@ class ScheduleFunctions extends ScheduleBase
 
                     $this->updateProjectAccess($schedule['project'], $dbForPlatform);
 
-                    $isRedisFallback = \str_contains(System::getEnv('_APP_WORKER_REDIS_FALLBACK', ''), 'functions');
-
-                    $queueForFunctions = new Func(
-                        $isRedisFallback
-                        ? $this->publisherRedis
-                        : $this->publisher
-                    );
+                    $queueForFunctions = new Func($this->publisher);
 
                     $queueForFunctions
                         ->setType('schedule')

src/Appwrite/Platform/Tasks/ScheduleMessages.php

@@ -4,7 +4,6 @@ namespace Appwrite\Platform\Tasks;
 
 use Appwrite\Event\Messaging;
 use Utopia\Database\Database;
-use Utopia\Pools\Group;
 
 class ScheduleMessages extends ScheduleBase
 {
@@ -26,7 +25,7 @@ class ScheduleMessages extends ScheduleBase
         return 'messages';
     }
 
-    protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
+    protected function enqueueResources(Database $dbForPlatform, callable $getProjectDB): void
     {
         foreach ($this->schedules as $schedule) {
             if (!$schedule['active']) {