adding realtime region sync

app/config/collections.php

@@ -23,6 +23,15 @@ $collections = [
         '$id' => ID::custom('syncs'),
         'name' => 'Syncs',
         'attributes' => [
+            [
+                '$id' => ID::custom('type'),
+                'type' => Database::VAR_STRING,
+                'size' => 256,
+                'required' => true,
+                'signed' => true,
+                'array' => false,
+                'filters' => [],
+            ],
             [
                 '$id' => ID::custom('region'),
                 'type' => Database::VAR_STRING,
@@ -42,7 +51,7 @@ $collections = [
                 'filters' => [],
             ],
             [
-                '$id' => ID::custom('keys'),
+                '$id' => ID::custom('key'),
                 'type' => Database::VAR_STRING,
                 'format' => '',
                 'size' => 16384,
@@ -61,23 +70,12 @@ $collections = [
                 'array' => false,
                 'filters' => [],
             ],
-            [
-                '$id' => ID::custom('payload'),
-                'type' => Database::VAR_STRING,
-                'format' => '',
-                'size' => 16384,
-                'signed' => true,
-                'required' => true,
-                'default' => [],
-                'array' => false,
-                'filters' => ['json'],
-            ],
         ],
         'indexes' => [
             [
-                '$id' => ID::custom('_key_status'),
+                '$id' => ID::custom('_key_type_status'),
                 'type' => Database::INDEX_KEY,
-                'attributes' => ['status'],
+                'attributes' => ['type', 'status'],
                 'lengths' => [],
                 'orders' => [],
             ],

app/controllers/api/edge.php

@@ -6,10 +6,13 @@ use Appwrite\Extend\Exception;
 use Appwrite\Utopia\Request;
 use Appwrite\Utopia\Response;
 use Utopia\App;
+use Utopia\Database\Database;
 use Utopia\Database\Document;
 use Utopia\Queue\Client;
 use Utopia\Validator\ArrayList;
+use Utopia\Validator\Assoc;
 use Utopia\Validator\Text;
+use Utopia\Validator\WhiteList;
 
 App::init()
     ->groups(['edge'])
@@ -30,20 +33,23 @@ App::post('/v1/edge/sync')
     ->desc('Purge cache keys')
     ->groups(['edge'])
     ->label('scope', 'public')
-    ->param('keys', '', new ArrayList(new Text(100), 1000), 'Cache keys. an array containing alphanumerical cache keys')
+    ->param('keys', '', new ArrayList(new Assoc(), 500), 'Cache keys. an array containing alphanumerical cache keys')
     ->inject('request')
     ->inject('response')
     ->inject('queueForCacheSyncIn')
     ->action(function (array $keys, Request $request, Response $response, Client $queueForCacheSyncIn) {
 
-        if (empty($keys)) {
+        //if (empty($keys)) {
             throw new Exception(Exception::KEY_NOT_FOUND);
-        }
+        //}
 
-        $queueForCacheSyncIn
-            ->enqueue([
-                'keys' => $keys
-            ]);
+        foreach ($keys as $sync) {
+            $queueForCacheSyncIn
+                ->enqueue([
+                    'type' => $sync['type'],
+                    'key'  => $sync['key']
+                ]);
+        }
 
         $response->dynamic(new Document([
             'keys' => $keys

app/controllers/shared/api.php

@@ -22,6 +22,7 @@ use Utopia\Database\Database;
 use Utopia\Database\DateTime;
 use Utopia\Database\Document;
 use Utopia\Database\Validator\Authorization;
+use Utopia\Queue\Client;
 
 $parseLabel = function (string $label, array $responsePayload, array $requestParams, Document $user) {
     preg_match_all('/{(.*?)}/', $label, $matches);
@@ -333,7 +334,8 @@ App::shutdown()
     ->inject('mode')
     ->inject('dbForProject')
     ->inject('queueForFunctions')
-    ->action(function (App $utopia, Request $request, Response $response, Document $project, Event $events, Audit $audits, Stats $usage, Delete $deletes, EventDatabase $database, string $mode, Database $dbForProject, Func $queueForFunctions) use ($parseLabel) {
+    ->inject('queueForCacheSyncOut')
+    ->action(function (App $utopia, Request $request, Response $response, Document $project, Event $events, Audit $audits, Stats $usage, Delete $deletes, EventDatabase $database, string $mode, Database $dbForProject, Func $queueForFunctions, Client $queueForCacheSyncOut) use ($parseLabel) {
 
         $responsePayload = $response->getPayload();
 
@@ -388,6 +390,21 @@ App::shutdown()
                         'userId' => $events->getParam('userId')
                     ]
                 );
+
+                $queueForCacheSyncOut->enqueue([
+                        'type' => 'realtime',
+                        'key' => [
+                                'projectId' => $target['projectId'] ?? $project->getId(),
+                                'payload' => $events->getPayload(),
+                                'events' => $allEvents,
+                                'channels' => $target['channels'],
+                                'roles' => $target['roles'],
+                                'options' => [
+                                    'permissionsChanged' => $target['permissionsChanged'],
+                                    'userId' => $events->getParam('userId')
+                                ]
+                        ]
+                ]);
             }
         }
 
@@ -511,6 +528,7 @@ App::shutdown()
 
             $fileSize = 0;
             $file = $request->getFiles('file');
+
             if (!empty($file)) {
                 $fileSize = (\is_array($file['size']) && isset($file['size'][0])) ? $file['size'][0] : $file['size'];
             }

docker-compose.yml

@@ -85,7 +85,7 @@ services:
       - ./public:/usr/src/code/public
       - ./src:/usr/src/code/src
       - ./dev:/usr/local/dev
-
+      - ./vendor/utopia-php/framework:/usr/src/code/vendor/utopia-php/framework
       #- ./vendor/utopia-php/cache:/usr/src/code/vendor/utopia-php/cache
     depends_on:
       - mariadb