sync against 1.6x

src/Appwrite/Event/Build.php

@@ -115,6 +115,7 @@ class Build extends Event
         $client = new Client($this->queue, $this->connection);
 
         return $client->enqueue([
+            'sourceRegion' =>  System::getEnv('_APP_REGION', 'default'),
             'project' => $this->project,
             'resource' => $this->resource,
             'deployment' => $this->deployment,

src/Appwrite/Event/Certificate.php

@@ -77,6 +77,7 @@ class Certificate extends Event
         $client = new Client($this->queue, $this->connection);
 
         return $client->enqueue([
+            'sourceRegion' =>  System::getEnv('_APP_REGION', 'default'),
             'project' => $this->project,
             'domain' => $this->domain,
             'skipRenewCheck' => $this->skipRenewCheck

src/Appwrite/Event/Database.php

@@ -121,6 +121,7 @@ class Database extends Event
 
         try {
             $result = $client->enqueue([
+                'sourceRegion' =>  System::getEnv('_APP_REGION', 'default'),
                 'project' => $this->project,
                 'user' => $this->user,
                 'type' => $this->type,

src/Appwrite/Event/Event.php

@@ -6,6 +6,7 @@ use InvalidArgumentException;
 use Utopia\Database\Document;
 use Utopia\Queue\Client;
 use Utopia\Queue\Connection;
+use Utopia\System\System;
 
 class Event
 {
@@ -109,6 +110,11 @@ class Event
         return $this->event;
     }
 
+    public function getSourceRegion(): string
+    {
+        return System::getEnv('_APP_REGION', 'default');
+    }
+
     /**
      * Set project for this event.
      *
@@ -301,6 +307,7 @@ class Event
         $client = new Client($this->queue, $this->connection);
 
         return $client->enqueue([
+            'sourceRegion' =>  $this->getSourceRegion(),
             'project' => $this->project,
             'user' => $this->user,
             'payload' => $this->payload,

src/Appwrite/Event/Migration.php

@@ -79,6 +79,8 @@ class Migration extends Event
         $client = new Client($this->queue, $this->connection);
 
         return $client->enqueue([
+            'sourceRegion' =>  $this->getSourceRegion(),
+
             'project' => $this->project,
             'user' => $this->user,
             'migration' => $this->migration

src/Appwrite/Messaging/Adapter.php

@@ -5,6 +5,8 @@ namespace Appwrite\Messaging;
 abstract class Adapter
 {
     abstract public function subscribe(string $projectId, mixed $identifier, array $roles, array $channels): void;
+
     abstract public function unsubscribe(mixed $identifier): void;
-    abstract public static function send(string $projectId, array $payload, array $events, array $channels, array $roles, array $options): void;
+
+    abstract public static function send(\redis $redis, string $projectId, array $payload, array $events, array $channels, array $roles, array $options): void;
 }

src/Appwrite/Messaging/Adapter/Realtime.php

@@ -122,15 +122,17 @@ class Realtime extends Adapter
 
     /**
      * Sends an event to the Realtime Server
+     * @param \Redis $redis
      * @param string $projectId
      * @param array $payload
-     * @param string $event
+     * @param array $events
      * @param array $channels
      * @param array $roles
      * @param array $options
      * @return void
+     * @throws \RedisException
      */
-    public static function send(string $projectId, array $payload, array $events, array $channels, array $roles, array $options = []): void
+    public static function send(\Redis $redis, string $projectId, array $payload, array $events, array $channels, array $roles, array $options = []): void
     {
         if (empty($channels) || empty($roles) || empty($projectId)) {
             return;
@@ -139,8 +141,6 @@ class Realtime extends Adapter
         $permissionsChanged = array_key_exists('permissionsChanged', $options) && $options['permissionsChanged'];
         $userId = array_key_exists('userId', $options) ? $options['userId'] : null;
 
-        $redis = new \Redis(); //TODO: make this part of the constructor
-        $redis->connect(System::getEnv('_APP_REDIS_HOST', ''), System::getEnv('_APP_REDIS_PORT', ''));
         $redis->publish('realtime', json_encode([
             'project' => $projectId,
             'roles' => $roles,

src/Appwrite/Platform/Workers/Builds.php

@@ -33,6 +33,11 @@ use Utopia\VCS\Adapter\Git\GitHub;
 
 class Builds extends Action
 {
+    /**
+     * @var mixed|string
+     */
+    protected string $sourceRegion;
+
     public static function getName(): string
     {
         return 'builds';
@@ -54,7 +59,8 @@ class Builds extends Action
             ->inject('dbForProject')
             ->inject('deviceForFunctions')
             ->inject('log')
-            ->callback(fn ($message, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions, Usage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log) => $this->action($message, $dbForConsole, $queueForEvents, $queueForFunctions, $usage, $cache, $dbForProject, $deviceForFunctions, $log));
+            ->inject('realtimeConnection')
+            ->callback(fn ($message, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions, Usage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log, Callable $realtimeConnection) => $this->action($message, $dbForConsole, $queueForEvents, $queueForFunctions, $usage, $cache, $dbForProject, $deviceForFunctions, $log, $realtimeConnection));
     }
 
     /**
@@ -67,10 +73,11 @@ class Builds extends Action
      * @param Database $dbForProject
      * @param Device $deviceForFunctions
      * @param Log $log
+     * @param callable $realtimeConnection
      * @return void
      * @throws \Utopia\Database\Exception
      */
-    public function action(Message $message, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions, Usage $queueForUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log): void
+    public function action(Message $message, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions, Usage $queueForUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, Log $log, Callable $realtimeConnection): void
     {
         $payload = $message->getPayload() ?? [];
 
@@ -84,6 +91,7 @@ class Builds extends Action
         $deployment = new Document($payload['deployment'] ?? []);
         $template = new Document($payload['template'] ?? []);
 
+        $this->sourceRegion = $payload['sourceRegion'] ?? 'default';
         $log->addTag('projectId', $project->getId());
         $log->addTag('type', $type);
 
@@ -92,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, $queueForUsage, $dbForConsole, $dbForProject, $github, $project, $resource, $deployment, $template, $log);
+                $this->buildDeployment($deviceForFunctions, $queueForFunctions, $queueForEvents, $queueForUsage, $dbForConsole, $dbForProject, $github, $project, $resource, $deployment, $template, $log, $realtimeConnection);
                 break;
 
             default:
@@ -117,7 +125,7 @@ class Builds extends Action
      * @throws \Utopia\Database\Exception
      * @throws Exception
      */
-    protected function buildDeployment(Device $deviceForFunctions, Func $queueForFunctions, Event $queueForEvents, Usage $queueForUsage, Database $dbForConsole, Database $dbForProject, GitHub $github, Document $project, Document $function, Document $deployment, Document $template, Log $log): void
+    protected function buildDeployment(Device $deviceForFunctions, Func $queueForFunctions, Event $queueForEvents, Usage $queueForUsage, Database $dbForConsole, Database $dbForProject, GitHub $github, Document $project, Document $function, Document $deployment, Document $template, Log $log, Callable $realtimeConnection): void
     {
         $executor = new Executor(System::getEnv('_APP_EXECUTOR_HOST'));
 
@@ -376,6 +384,7 @@ class Builds extends Action
                         project: $project
                     );
                     Realtime::send(
+                        redis: $realtimeConnection($this->sourceRegion),
                         projectId: 'console',
                         payload: $build->getArrayCopy(),
                         events: $allEvents,
@@ -454,6 +463,7 @@ class Builds extends Action
             );
 
             Realtime::send(
+                redis: $realtimeConnection($this->sourceRegion),
                 projectId: 'console',
                 payload: $build->getArrayCopy(),
                 events: $allEvents,
@@ -552,12 +562,12 @@ class Builds extends Action
                         $err = $error;
                     }
                 }),
-                Co\go(function () use ($executor, $project, $deployment, &$response, &$build, $dbForProject, $allEvents, &$err, &$isCanceled) {
+                Co\go(function () use ($realtimeConnection, $executor, $project, $deployment, &$response, &$build, $dbForProject, $allEvents, &$err, &$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 ($realtimeConnection, &$response, &$err, &$build, $dbForProject, $allEvents, $project, &$isCanceled) {
                                 if ($isCanceled) {
                                     return;
                                 }
@@ -591,6 +601,7 @@ class Builds extends Action
                                         project: $project
                                     );
                                     Realtime::send(
+                                        redis: $realtimeConnection($this->sourceRegion),
                                         projectId: 'console',
                                         payload: $build->getArrayCopy(),
                                         events: $allEvents,
@@ -693,6 +704,7 @@ class Builds extends Action
                 project: $project
             );
             Realtime::send(
+                redis: $realtimeConnection($this->sourceRegion),
                 projectId: 'console',
                 payload: $build->getArrayCopy(),
                 events: $allEvents,

src/Appwrite/Platform/Workers/Certificates.php

@@ -30,6 +30,11 @@ use Utopia\System\System;
 
 class Certificates extends Action
 {
+    /**
+     * @var mixed|string
+     */
+    protected string $sourceRegion;
+
     public static function getName(): string
     {
         return 'certificates';
@@ -48,7 +53,8 @@ class Certificates extends Action
             ->inject('queueForEvents')
             ->inject('queueForFunctions')
             ->inject('log')
-            ->callback(fn (Message $message, Database $dbForConsole, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log) => $this->action($message, $dbForConsole, $queueForMails, $queueForEvents, $queueForFunctions, $log));
+            ->inject('realtimeConnection')
+            ->callback(fn (Message $message, Database $dbForConsole, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, Callable $realtimeConnection) => $this->action($message, $dbForConsole, $queueForMails, $queueForEvents, $queueForFunctions, $log, $realtimeConnection));
     }
 
     /**
@@ -58,11 +64,12 @@ class Certificates extends Action
      * @param Event $queueForEvents
      * @param Func $queueForFunctions
      * @param Log $log
+     * @param callable $realtimeConnection
      * @return void
      * @throws Throwable
      * @throws \Utopia\Database\Exception
      */
-    public function action(Message $message, Database $dbForConsole, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log): void
+    public function action(Message $message, Database $dbForConsole, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, Callable $realtimeConnection): void
     {
         $payload = $message->getPayload() ?? [];
 
@@ -74,9 +81,9 @@ class Certificates extends Action
         $domain   = new Domain($document->getAttribute('domain', ''));
         $skipRenewCheck = $payload['skipRenewCheck'] ?? false;
 
+        $this->sourceRegion = $payload['sourceRegion'] ?? 'default';
         $log->addTag('domain', $domain->get());
-
-        $this->execute($domain, $dbForConsole, $queueForMails, $queueForEvents, $queueForFunctions, $log, $skipRenewCheck);
+        $this->execute($domain, $dbForConsole, $queueForMails, $queueForEvents, $queueForFunctions, $log,$realtimeConnection, $skipRenewCheck);
     }
 
     /**
@@ -85,12 +92,17 @@ class Certificates extends Action
      * @param Mail $queueForMails
      * @param Event $queueForEvents
      * @param Func $queueForFunctions
+     * @param Log $log
+     * @param callable $realtimeConnection
      * @param bool $skipRenewCheck
      * @return void
+     * @throws Authorization
+     * @throws Conflict
+     * @throws Structure
      * @throws Throwable
      * @throws \Utopia\Database\Exception
      */
-    private function execute(Domain $domain, Database $dbForConsole, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, bool $skipRenewCheck = false): void
+    protected function execute(Domain $domain, Database $dbForConsole, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, Callable $realtimeConnection, bool $skipRenewCheck = false): void
     {
         /**
          * 1. Read arguments and validate domain
@@ -193,7 +205,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, $dbForConsole, $queueForEvents, $queueForFunctions);
+            $this->saveCertificateDocument($domain->get(), $certificate, $success, $dbForConsole, $queueForEvents, $queueForFunctions, $realtimeConnection);
         }
     }
 
@@ -212,7 +224,7 @@ class Certificates extends Action
      * @throws Conflict
      * @throws Structure
      */
-    private function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions): void
+    protected function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions, Callable $realtimeConnection): void
     {
         // Check if update or insert required
         $certificateDocument = $dbForConsole->findOne('certificates', [Query::equal('domain', [$domain])]);
@@ -226,7 +238,7 @@ class Certificates extends Action
         }
 
         $certificateId = $certificate->getId();
-        $this->updateDomainDocuments($certificateId, $domain, $success, $dbForConsole, $queueForEvents, $queueForFunctions);
+        $this->updateDomainDocuments($certificateId, $domain, $success, $dbForConsole, $queueForEvents, $queueForFunctions, $realtimeConnection);
     }
 
     /**
@@ -234,7 +246,7 @@ class Certificates extends Action
      *
      * @return null|string Returns main domain. If null, there is no main domain yet.
      */
-    private function getMainDomain(): ?string
+    protected function getMainDomain(): ?string
     {
         $envDomain = System::getEnv('_APP_DOMAIN', '');
         if (!empty($envDomain) && $envDomain !== 'localhost') {
@@ -255,7 +267,7 @@ class Certificates extends Action
      * @return void
      * @throws Exception
      */
-    private function validateDomain(Domain $domain, bool $isMainDomain, Log $log): void
+    protected function validateDomain(Domain $domain, bool $isMainDomain, Log $log): void
     {
         if (empty($domain->get())) {
             throw new Exception('Missing certificate domain.');
@@ -299,7 +311,7 @@ class Certificates extends Action
      * @return bool True, if certificate needs to be renewed
      * @throws Exception
      */
-    private function isRenewRequired(string $domain, Log $log): bool
+    protected function isRenewRequired(string $domain, Log $log): bool
     {
         $certPath = APP_STORAGE_CERTIFICATES . '/' . $domain . '/cert.pem';
         if (\file_exists($certPath)) {
@@ -333,7 +345,7 @@ class Certificates extends Action
      * @return array Named array with keys 'stdout' and 'stderr', both string
      * @throws Exception
      */
-    private function issueCertificate(string $folder, string $domain, string $email): array
+    protected function issueCertificate(string $folder, string $domain, string $email): array
     {
         $stdout = '';
         $stderr = '';
@@ -363,7 +375,7 @@ class Certificates extends Action
      * @return string
      * @throws \Utopia\Database\Exception
      */
-    private function getRenewDate(string $domain): string
+    protected function getRenewDate(string $domain): string
     {
         $certPath = APP_STORAGE_CERTIFICATES . '/' . $domain . '/cert.pem';
         $certData = openssl_x509_parse(file_get_contents($certPath));
@@ -381,7 +393,7 @@ class Certificates extends Action
      * @return void
      * @throws Exception
      */
-    private function applyCertificateFiles(string $folder, string $domain, array $letsEncryptData): void
+    protected function applyCertificateFiles(string $folder, string $domain, array $letsEncryptData): void
     {
 
         // Prepare folder in storage for domain
@@ -432,7 +444,7 @@ class Certificates extends Action
      * @return void
      * @throws Exception
      */
-    private function notifyError(string $domain, string $errorMessage, int $attempt, Mail $queueForMails): void
+    protected function notifyError(string $domain, string $errorMessage, int $attempt, Mail $queueForMails): void
     {
         // Log error into console
         Console::warning('Cannot renew domain (' . $domain . ') on attempt no. ' . $attempt . ' certificate: ' . $errorMessage);
@@ -475,7 +487,7 @@ class Certificates extends Action
      *
      * @return void
      */
-    private function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions): void
+    protected function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForConsole, Event $queueForEvents, Func $queueForFunctions, Callable $realtimeConnection): void
     {
 
         $rule = $dbForConsole->findOne('rules', [
@@ -525,6 +537,7 @@ class Certificates extends Action
                 project: $project
             );
             Realtime::send(
+                redis: $realtimeConnection($this->sourceRegion),
                 projectId: 'console',
                 payload: $rule->getArrayCopy(),
                 events: $allEvents,
@@ -532,6 +545,7 @@ class Certificates extends Action
                 roles: $target['roles']
             );
             Realtime::send(
+                redis: $realtimeConnection($this->sourceRegion),
                 projectId: $project->getId(),
                 payload: $rule->getArrayCopy(),
                 events: $allEvents,

src/Appwrite/Platform/Workers/Databases.php

@@ -21,6 +21,11 @@ use Utopia\Queue\Message;
 
 class Databases extends Action
 {
+    /**
+     * @var array|mixed
+     */
+    protected string $sourceRegion;
+
     public static function getName(): string
     {
         return 'databases';
@@ -37,7 +42,8 @@ class Databases extends Action
             ->inject('dbForConsole')
             ->inject('dbForProject')
             ->inject('log')
-            ->callback(fn (Message $message, Database $dbForConsole, Database $dbForProject, Log $log) => $this->action($message, $dbForConsole, $dbForProject, $log));
+            ->inject('realtimeConnection')
+            ->callback(fn (Message $message, Database $dbForConsole, Database $dbForProject, Log $log, callable $realtimeConnection) => $this->action($message, $dbForConsole, $dbForProject, $log, $realtimeConnection));
     }
 
     /**
@@ -45,10 +51,16 @@ class Databases extends Action
      * @param Database $dbForConsole
      * @param Database $dbForProject
      * @param Log $log
+     * @param callable $realtimeConnection
      * @return void
-     * @throws \Exception
+     * @throws Authorization
+     * @throws Conflict
+     * @throws DatabaseException
+     * @throws Restricted
+     * @throws Structure
+     * @throws Exception
      */
-    public function action(Message $message, Database $dbForConsole, Database $dbForProject, Log $log): void
+    public function action(Message $message, Database $dbForConsole, Database $dbForProject, Log $log, callable $realtimeConnection): void
     {
         $payload = $message->getPayload() ?? [];
 
@@ -62,6 +74,7 @@ class Databases extends Action
         $document = new Document($payload['document'] ?? []);
         $database = new Document($payload['database'] ?? []);
 
+        $this->sourceRegion = $payload['sourceRegion'] ?? 'default';
         $log->addTag('projectId', $project->getId());
         $log->addTag('type', $type);
 
@@ -74,10 +87,10 @@ 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, $dbForConsole, $dbForProject),
-            DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForConsole, $dbForProject),
-            DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForConsole, $dbForProject),
-            DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForConsole, $dbForProject),
+            DATABASE_TYPE_CREATE_ATTRIBUTE => $this->createAttribute($database, $collection, $document, $project, $dbForConsole, $dbForProject, $realtimeConnection),
+            DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForConsole, $dbForProject, $realtimeConnection),
+            DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForConsole, $dbForProject, $realtimeConnection),
+            DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForConsole, $dbForProject, $realtimeConnection),
             default => throw new \Exception('No database operation for type: ' . \strval($type)),
         };
     }
@@ -89,12 +102,15 @@ class Databases extends Action
      * @param Document $project
      * @param Database $dbForConsole
      * @param Database $dbForProject
+     * @param callable $realtimeConnection
      * @return void
      * @throws Authorization
      * @throws Conflict
-     * @throws \Exception
+     * @throws DatabaseException
+     * @throws Structure
+     * @throws Exception
      */
-    private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForConsole, Database $dbForProject): void
+    private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForConsole, Database $dbForProject, callable $realtimeConnection): void
     {
         if ($collection->isEmpty()) {
             throw new Exception('Missing collection');
@@ -195,7 +211,7 @@ class Databases extends Action
                 );
             }
         } finally {
-            $this->trigger($database, $collection, $attribute, $project, $projectId, $events);
+            $this->trigger($database, $collection, $attribute, $project, $projectId, $events, $realtimeConnection);
         }
 
         if ($type === Database::VAR_RELATIONSHIP && $options['twoWay']) {
@@ -212,12 +228,16 @@ class Databases extends Action
      * @param Document $project
      * @param Database $dbForConsole
      * @param Database $dbForProject
+     * @param callable $realtimeConnection
      * @return void
      * @throws Authorization
      * @throws Conflict
-     * @throws \Exception
-     **/
-    private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForConsole, Database $dbForProject): void
+     * @throws DatabaseException
+     * @throws Restricted
+     * @throws Structure
+     * @throws Exception
+     */
+    private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForConsole, Database $dbForProject, callable $realtimeConnection): void
     {
         if ($collection->isEmpty()) {
             throw new Exception('Missing collection');
@@ -296,7 +316,7 @@ class Databases extends Action
                 );
             }
         } finally {
-            $this->trigger($database, $collection, $attribute, $project, $projectId, $events);
+            $this->trigger($database, $collection, $attribute, $project, $projectId, $events, $realtimeConnection);
         }
 
         // The underlying database removes/rebuilds indexes when attribute is removed
@@ -366,13 +386,15 @@ class Databases extends Action
      * @param Document $project
      * @param Database $dbForConsole
      * @param Database $dbForProject
+     * @param callable $realtimeConnection
      * @return void
      * @throws Authorization
      * @throws Conflict
-     * @throws Structure
      * @throws DatabaseException
+     * @throws Structure
+     * @throws Exception
      */
-    private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForConsole, Database $dbForProject): void
+    private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForConsole, Database $dbForProject, callable $realtimeConnection): void
     {
         if ($collection->isEmpty()) {
             throw new Exception('Missing collection');
@@ -414,7 +436,7 @@ class Databases extends Action
                 $index->setAttribute('status', 'failed')
             );
         } finally {
-            $this->trigger($database, $collection, $index, $project, $projectId, $events);
+            $this->trigger($database, $collection, $index, $project, $projectId, $events, $realtimeConnection);
         }
 
         $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collectionId);
@@ -427,13 +449,15 @@ class Databases extends Action
      * @param Document $project
      * @param Database $dbForConsole
      * @param Database $dbForProject
+     * @param callable $realtimeConnection
      * @return void
      * @throws Authorization
      * @throws Conflict
-     * @throws Structure
      * @throws DatabaseException
+     * @throws Structure
+     * @throws Exception
      */
-    private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForConsole, Database $dbForProject): void
+    private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForConsole, Database $dbForProject, callable $realtimeConnection): void
     {
         if ($collection->isEmpty()) {
             throw new Exception('Missing collection');
@@ -472,7 +496,7 @@ class Databases extends Action
                 $index->setAttribute('status', 'stuck')
             );
         } finally {
-            $this->trigger($database, $collection, $index, $project, $projectId, $events);
+            $this->trigger($database, $collection, $index, $project, $projectId, $events, $realtimeConnection);
         }
 
         $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collection->getId());
@@ -612,13 +636,18 @@ class Databases extends Action
         Console::info("Deleted {$count} document by group in " . ($executionEnd - $executionStart) . " seconds");
     }
 
+    /**
+     * @throws \RedisException
+     * @throws Exception
+     */
     protected function trigger(
         Document $database,
         Document $collection,
         Document $attribute,
         Document $project,
         string $projectId,
-        array $events
+        array $events,
+        callable $realtimeConnection
     ): void {
         $target = Realtime::fromPayload(
             // Pass first, most verbose event pattern
@@ -627,6 +656,7 @@ class Databases extends Action
             project: $project,
         );
         Realtime::send(
+            redis: $realtimeConnection($this->sourceRegion),
             projectId: 'console',
             payload: $attribute->getArrayCopy(),
             events: $events,

src/Appwrite/Platform/Workers/Deletes.php

@@ -497,19 +497,20 @@ class Deletes extends Action
         ];
 
         $limit = \count($projectCollectionIds) + 25;
+        $sharedTablesKeys = explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', ''));
 
         while (true) {
             $collections = $dbForProject->listCollections($limit);
 
             foreach ($collections as $collection) {
-                if ($dsn->getHost() !== System::getEnv('_APP_DATABASE_SHARED_TABLES', '') || !\in_array($collection->getId(), $projectCollectionIds)) {
-                    $dbForProject->deleteCollection($collection->getId());
+                if (! in_array($dsn->getHost(), $sharedTablesKeys) || !\in_array($collection->getId(), $projectCollectionIds)) {
+                    $dbForProject->deleteCollection($collection->getId());                    $dbForProject->deleteCollection($collection->getId());
                 } else {
                     $this->deleteByGroup($collection->getId(), [], database: $dbForProject);
                 }
             }
 
-            if ($dsn->getHost() === System::getEnv('_APP_DATABASE_SHARED_TABLES', '')) {
+            if (in_array($dsn->getHost(), $sharedTablesKeys)) {
                 $collectionsIds = \array_map(fn ($collection) => $collection->getId(), $collections);
 
                 if (empty(\array_diff($collectionsIds, $projectCollectionIds))) {
@@ -558,7 +559,8 @@ class Deletes extends Action
         ], $dbForConsole);
 
         // Delete metadata table
-        if ($dsn->getHost() !== System::getEnv('_APP_DATABASE_SHARED_TABLES', '')) {
+        System::getEnv('_APP_DATABASE_SHARED_TABLES', '');
+        if (! in_array($dsn, $sharedTablesKeys)) {
             $dbForProject->deleteCollection('_metadata');
         } else {
             $this->deleteByGroup('_metadata', [], $dbForProject);

src/Appwrite/Platform/Workers/Functions.php

@@ -28,6 +28,11 @@ use Utopia\System\System;
 
 class Functions extends Action
 {
+    /**
+     * @var mixed|string
+     */
+    protected string $sourceRegion;
+
     public static function getName(): string
     {
         return 'functions';
@@ -47,7 +52,8 @@ class Functions extends Action
             ->inject('queueForEvents')
             ->inject('queueForUsage')
             ->inject('log')
-            ->callback(fn (Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, Usage $queueForUsage, Log $log) => $this->action($message, $dbForProject, $queueForFunctions, $queueForEvents, $queueForUsage, $log));
+            ->inject('realtimeConnection')
+            ->callback(fn (Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, Usage $queueForUsage, Log $log, Callable $realtimeConnection) => $this->action($message, $dbForProject, $queueForFunctions, $queueForEvents, $queueForUsage, $log, $realtimeConnection));
     }
 
     /**
@@ -63,7 +69,7 @@ class Functions extends Action
      * @throws \Utopia\Database\Exception
      * @throws Conflict
      */
-    public function action(Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, Usage $queueForUsage, Log $log): void
+    public function action(Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, Usage $queueForUsage, Log $log, Callable $realtimeConnection): void
     {
         $payload = $message->getPayload() ?? [];
 
@@ -93,6 +99,9 @@ class Functions extends Action
             return;
         }
 
+        $this->sourceRegion = $payload['sourceRegion'] ?? 'default';
+
+
         if ($function->isEmpty() && !empty($functionId)) {
             $function = $dbForProject->getDocument('functions', $functionId);
         }
@@ -125,6 +134,7 @@ class Functions extends Action
                     Console::success('Iterating function: ' . $function->getAttribute('name'));
 
                     $this->execute(
+                        realtimeConnection: $realtimeConnection,
                         log: $log,
                         dbForProject: $dbForProject,
                         queueForFunctions: $queueForFunctions,
@@ -161,6 +171,7 @@ class Functions extends Action
                 $execution = new Document($payload['execution'] ?? []);
                 $user = new Document($payload['user'] ?? []);
                 $this->execute(
+                    realtimeConnection: $realtimeConnection,
                     log: $log,
                     dbForProject: $dbForProject,
                     queueForFunctions: $queueForFunctions,
@@ -183,6 +194,7 @@ class Functions extends Action
             case 'schedule':
                 $execution = new Document($payload['execution'] ?? []);
                 $this->execute(
+                    realtimeConnection: $realtimeConnection,
                     log: $log,
                     dbForProject: $dbForProject,
                     queueForFunctions: $queueForFunctions,
@@ -292,6 +304,7 @@ class Functions extends Action
      * @throws Conflict
      */
     private function execute(
+        Callable $realtimeConnection,
         Log $log,
         Database $dbForProject,
         Func $queueForFunctions,
@@ -583,6 +596,7 @@ class Functions extends Action
             payload: $execution
         );
         Realtime::send(
+            redis: $realtimeConnection($this->sourceRegion),
             projectId: 'console',
             payload: $execution->getArrayCopy(),
             events: $allEvents,
@@ -590,6 +604,7 @@ class Functions extends Action
             roles: $target['roles']
         );
         Realtime::send(
+            redis: $realtimeConnection($this->sourceRegion),
             projectId: $project->getId(),
             payload: $execution->getArrayCopy(),
             events: $allEvents,

src/Appwrite/Platform/Workers/Migrations.php

@@ -32,6 +32,10 @@ class Migrations extends Action
 {
     private ?Database $dbForProject = null;
     private ?Database $dbForConsole = null;
+    /**
+     * @var string
+     */
+    protected string $sourceRegion;
 
     public static function getName(): string
     {
@@ -49,7 +53,8 @@ class Migrations extends Action
             ->inject('dbForProject')
             ->inject('dbForConsole')
             ->inject('log')
-            ->callback(fn (Message $message, Database $dbForProject, Database $dbForConsole, Log $log) => $this->action($message, $dbForProject, $dbForConsole, $log));
+            ->inject('realtimeConnection')
+            ->callback(fn (Message $message, Database $dbForProject, Database $dbForConsole, Log $log, Callable $realtimeConnection) => $this->action($message, $dbForProject, $dbForConsole, $log, $realtimeConnection));
     }
 
     /**
@@ -60,7 +65,7 @@ class Migrations extends Action
      * @return void
      * @throws Exception
      */
-    public function action(Message $message, Database $dbForProject, Database $dbForConsole, Log $log): void
+    public function action(Message $message, Database $dbForProject, Database $dbForConsole, Log $log,  Callable $realtimeConnection): void
     {
         $payload = $message->getPayload() ?? [];
 
@@ -76,6 +81,7 @@ class Migrations extends Action
             return;
         }
 
+        $this->sourceRegion = $payload['sourceRegion'] ?? 'default';
         $this->dbForProject = $dbForProject;
         $this->dbForConsole = $dbForConsole;
 
@@ -89,7 +95,7 @@ class Migrations extends Action
         $log->addTag('migrationId', $migration->getId());
         $log->addTag('projectId', $project->getId());
 
-        $this->processMigration($project, $migration, $log);
+        $this->processMigration($project, $migration, $log, $realtimeConnection);
     }
 
     /**
@@ -134,7 +140,7 @@ 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, Callable $realtimeConnection): Document
     {
         /** Trigger Realtime */
         $allEvents = Event::generateEvents('migrations.[migrationId].update', [
@@ -148,6 +154,7 @@ class Migrations extends Action
         );
 
         Realtime::send(
+            redis: $realtimeConnection($this->sourceRegion),
             projectId: 'console',
             payload: $migration->getArrayCopy(),
             events: $allEvents,
@@ -156,6 +163,7 @@ class Migrations extends Action
         );
 
         Realtime::send(
+            redis: $realtimeConnection($this->sourceRegion),
             projectId: $project->getId(),
             payload: $migration->getArrayCopy(),
             events: $allEvents,
@@ -236,6 +244,7 @@ class Migrations extends Action
      * @param Document $project
      * @param Document $migration
      * @param Log $log
+     * @param callable $realtimeConnection
      * @return void
      * @throws Authorization
      * @throws Conflict
@@ -243,7 +252,7 @@ class Migrations extends Action
      * @throws Structure
      * @throws \Utopia\Database\Exception
      */
-    protected function processMigration(Document $project, Document $migration, Log $log): void
+    protected function processMigration(Document $project, Document $migration, Log $log, Callable $realtimeConnection): void
     {
         /**
          * @var Document $migrationDocument
@@ -259,7 +268,7 @@ class Migrations extends Action
             $migrationDocument->setAttribute('stage', 'processing');
             $migrationDocument->setAttribute('status', 'processing');
             $log->addBreadcrumb(new Breadcrumb("debug", "migration", "Migration hit stage 'processing'", \microtime(true)));
-            $this->updateMigrationDocument($migrationDocument, $projectDocument);
+            $this->updateMigrationDocument($migrationDocument, $projectDocument, $realtimeConnection);
 
             $log->addTag('type', $migrationDocument->getAttribute('source'));
 
@@ -281,12 +290,12 @@ class Migrations extends Action
             /** Start Transfer */
             $migrationDocument->setAttribute('stage', 'migrating');
             $log->addBreadcrumb(new Breadcrumb("debug", "migration", "Migration hit stage 'migrating'", \microtime(true)));
-            $this->updateMigrationDocument($migrationDocument, $projectDocument);
-            $transfer->run($migrationDocument->getAttribute('resources'), function () use ($migrationDocument, $transfer, $projectDocument) {
+            $this->updateMigrationDocument($migrationDocument, $projectDocument, $realtimeConnection);
+            $transfer->run($migrationDocument->getAttribute('resources'), function () use ($realtimeConnection, $migrationDocument, $transfer, $projectDocument) {
                 $migrationDocument->setAttribute('resourceData', json_encode($transfer->getCache()));
                 $migrationDocument->setAttribute('statusCounters', json_encode($transfer->getStatusCounters()));
 
-                $this->updateMigrationDocument($migrationDocument, $projectDocument);
+                $this->updateMigrationDocument($migrationDocument, $projectDocument, $realtimeConnection);
             });
 
             $sourceErrors = $source->getErrors();
@@ -309,7 +318,7 @@ class Migrations extends Action
 
                 $migrationDocument->setAttribute('errors', $errorMessages);
                 $log->addExtra('migrationErrors', json_encode($errorMessages));
-                $this->updateMigrationDocument($migrationDocument, $projectDocument);
+                $this->updateMigrationDocument($migrationDocument, $projectDocument, $realtimeConnection);
 
                 return;
             }
@@ -352,7 +361,7 @@ class Migrations extends Action
                 $this->removeAPIKey($tempAPIKey);
             }
             if ($migrationDocument) {
-                $this->updateMigrationDocument($migrationDocument, $projectDocument);
+                $this->updateMigrationDocument($migrationDocument, $projectDocument, $realtimeConnection);
 
                 if ($migrationDocument->getAttribute('status', '') == 'failed') {
                     throw new Exception("Migration failed");