adding realtime region sync

app/controllers/api/edge.php

@@ -39,9 +39,9 @@ App::post('/v1/edge/sync')
     ->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);
-        //}
+        }
 
         foreach ($keys as $sync) {
             $queueForCacheSyncIn

app/workers/sync-out.php

@@ -35,9 +35,9 @@ const MAX_CURL_SEND_ATTEMPTS = 4;
  * @param string $url
  * @param string $token
  * @param array $payload
- * @return array
+ * @return int
  */
-function call(string $url, string $token, array $payload): array
+function call(string $url, string $token, array $payload): int
 {
 
     $ch = curl_init($url);
@@ -82,12 +82,13 @@ function handle($dbForConsole, $regions, $payload): void
         $status = call($region['domain'] . '/v1/edge/sync', $token, ['keys' => $payload]);
         if ($status !== Response::STATUS_CODE_OK) {
             Console::error("[{$time}] Request to {$code} has failed");
+
             foreach ($payload as $sync) {
                 $dbForConsole->createDocument('syncs', new Document([
                     'region' => App::getEnv('_APP_REGION'),
                     'target' => $code,
                     'type' => $sync['type'],
-                    'key' => $sync['key'],
+                    'key'  => ['key' => $sync['key']],
                     'status' => $status,
                 ]));
             }
@@ -100,11 +101,6 @@ $server->job()
     ->action(function (Message $message) use (&$stack, &$failures) {
 
         $payload = $message->getPayload() ?? [];
-        $type    = $payload['type'] ?? null;
-
-        if (empty($type)) {
-            return;
-        }
 
         //Get failed requests
         if (!empty($payload['region']) && !empty($payload['keys'])) {
@@ -122,6 +118,10 @@ $server->job()
             return;
         }
 
+        if (empty($payload['type'])) {
+            return;
+        }
+
         if (!empty($payload['key'])) {
                 $stack['keys'][] = [
                     'type' => $payload['type'],
@@ -154,9 +154,8 @@ $server
                 }
                 return;
             }
-            //var_dump($stack['keys']);
+
             $chunk = array_slice($stack['keys'], 0, CHUNK_MAX_KEYS, true);
-            //var_dump($chunk);
             array_splice($stack['keys'], 0, CHUNK_MAX_KEYS);
             Console::log("[{$time}] Sending " . count($chunk) . " remains " . count($stack['keys']));
             handle($dbForConsole, $stack['regions'], $chunk);

src/Appwrite/Platform/Tasks/EdgeSync.php

@@ -47,7 +47,7 @@ class EdgeSync extends Action
 
             Console::success("[{$time}] New task every {$interval} seconds");
 
-            foreach ($regions as $target) {
+            foreach ($regions as $code => $region) {
                 $count = 0;
                 $chunk = 0;
                 $limit = 50;
@@ -58,16 +58,17 @@ class EdgeSync extends Action
 
                     $results = $dbForConsole->find('syncs', [
                         Query::equal('region', [App::getEnv('_APP_REGION')]),
-                        Query::equal('target', [$target]),
+                        Query::equal('target', [$code]),
                         Query::limit($limit)
                     ]);
 
                     $sum = count($results);
                     if ($sum > 0) {
                         foreach ($results as $document) {
+                            $key = $document->getAttribute('key');
                             $keys[] = [
                                         'type' => $document->getAttribute('type'),
-                                        'key' => $document->getAttribute('key')
+                                        'key'  => $key['key']
                                 ];
                             $dbForConsole->deleteDocument('syncs', $document->getId());
                             $count++;
@@ -76,14 +77,14 @@ class EdgeSync extends Action
                 }
 
                 if (!empty($keys)) {
-                    Console::info("[{$time}] Enqueueing  keys chunk {$count} to {$target}");
+                    Console::info("[{$time}] Enqueueing  keys chunk {$count} to region {$code}");
                     $queueForCacheSyncOut
                         ->enqueue([
-                            'region' => $target,
+                            'region' => $code,
                             'keys' => $keys
                         ]);
                 } else {
-                        Console::info("[{$time}] No cache keys where found.");
+                        Console::info("[{$time}] No  keys where found for  region {$code}.");
                 }
             }
         }, $interval);