From 601569805f60eecb6ac1e8de301a4c8da12031dc Mon Sep 17 00:00:00 2001
From: shimon <shimon@appwrite.io>
Date: Thu, 27 Oct 2022 17:01:42 +0300
Subject: [PATCH] Timer::tick

---
 app/controllers/api/edge.php |  4 +--
 app/tasks/sync-edge.php      |  2 +-
 app/workers/sync-out.php     | 48 ++++++++++++++++++++++--------------
 3 files changed, 32 insertions(+), 22 deletions(-)

diff --git a/app/controllers/api/edge.php b/app/controllers/api/edge.php
index bd80b91843..22c845a557 100644
--- a/app/controllers/api/edge.php
+++ b/app/controllers/api/edge.php
@@ -37,9 +37,9 @@ App::post('/v1/edge/sync')
     ->inject('response')
     ->action(function (array $keys, Request $request, Response $response) {
 
-        //if (empty($keys)) {
+        if (empty($keys)) {
             throw new Exception(Exception::KEY_NOT_FOUND);
-        //}
+        }
 
         $client = new SyncIn('syncIn', new QueueRedis(App::getEnv('_APP_REDIS_HOST', 'redis'), App::getEnv('_APP_REDIS_PORT', '6379')));
 
diff --git a/app/tasks/sync-edge.php b/app/tasks/sync-edge.php
index b9530b2d84..a3564a3685 100644
--- a/app/tasks/sync-edge.php
+++ b/app/tasks/sync-edge.php
@@ -16,7 +16,7 @@ $cli
     ->action(function () use ($register) {
         Console::title('Syncs edges V1');
         Console::success(APP_NAME . ' Syncs Edge process v1 has started');
-
+        sleep(3);
         $interval = (int) App::getEnv('_APP_SYNC_EDGE_INTERVAL', '180');
           Console::loop(function () use ($interval, $register) {
             $database = getConsoleDB();
diff --git a/app/workers/sync-out.php b/app/workers/sync-out.php
index b8dc9d8e2e..34aae45e34 100644
--- a/app/workers/sync-out.php
+++ b/app/workers/sync-out.php
@@ -9,13 +9,13 @@ use Swoole\Timer;
 use Utopia\App;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
+use Utopia\Database\DateTime;
 use Utopia\Database\Document;
 use Utopia\Database\Exception\Authorization;
 use Utopia\Database\Exception\Structure;
 use Utopia\Queue;
 use Utopia\Queue\Message;
 
-
 $regions = array_filter(
     Config::getParam('regions', []),
     fn ($region) => App::getEnv('_APP_REGION', 'nyc1') !== $region
@@ -27,10 +27,9 @@ $stack = [
     'regions' => $regions,
     'keys' => [],
 ];
-
 $failures = [];
 
-const MAX_KEY_COUNT = 2;
+const CHUNK_MAX_KEYS = 2;
 const MAX_CURL_SEND_ATTEMPTS = 4;
 
 /**
@@ -77,23 +76,31 @@ function send(string $url, string $token, array $stack): array
  * @throws Structure
  * @throws Exception
  */
-function call($database, $regions, $stack): void
+function call($regions, $stack): void
 {
 
+    global $register;
+
     $jwt = new JWT(App::getEnv('_APP_OPENSSL_KEY_V1'), 'HS256', 600, 10);
     $token = $jwt->encode([]);
 
     foreach ($regions as $code => $region) {
-        Console::info("Sending request to  {$code}");
+        $time = DateTime::now();
         $response = send($region['domain'] . '/v1/edge/sync', $token, ['keys' => $stack]);
         if ($response['status'] !== Response::STATUS_CODE_OK) {
-            $database->createDocument('syncs', new Document([
-                'region' => App::getEnv('_APP_REGION', 'nyc1'),
-                'target' => $code,
-                'keys' => $stack,
-                'status' => $response['status'],
-                'payload' => $response['payload'],
-            ]));
+            Console::error("[{$time}] Request to  {$code} has failed");
+            try {
+                $database = getConsoleDB();
+                $database->createDocument('syncs', new Document([
+                    'region' => App::getEnv('_APP_REGION', 'nyc1'),
+                    'target' => $code,
+                    'keys' => $stack,
+                    'status' => $response['status'],
+                    'payload' => $response['payload'],
+                ]));
+            } catch (\Throwable $th) {
+                $register->get('pools')->reclaim();
+            }
         }
     }
 }
@@ -108,7 +115,7 @@ $server->job()
     ->action(function (Message $message) use (&$stack, &$failures) {
 
         $payload = $message->getPayload()['value'] ?? [];
-
+        var_dump($message->getPayload());
         if (!empty($payload['keys'])) {
             $regions = array_filter(
                 Config::getParam('regions', []),
@@ -141,9 +148,10 @@ $server
 
 $server
     ->workerStart(function () use (&$stack, &$failures) {
-        Timer::tick(10000, function () use (&$stack, &$failures) {
+        Timer::tick(1000, function () use (&$stack, &$failures) {
+            $time = DateTime::now();
             if (empty($stack['keys']) && count($failures) === 0) {
-                Console::info("Stack is empty");
+                Console::info("[{$time}] Stack is empty");
                 return;
             }
 
@@ -151,15 +159,17 @@ $server
                 $i = 0;
                 while ($i < count($failures)) {
                     $failure = array_shift($failures);
-                    call(getConsoleDB(), $failure['regions'], $failure['keys']);
+                    Console::info("[{$time}] ReSending " . count($failure['keys']) . " to " . key($failure['regions']));
+                    call($failure['regions'], $failure['keys']);
                     $i++;
                 }
                 return;
             }
 
-            $chunk = array_slice($stack['keys'], 0, MAX_KEY_COUNT);
-            array_splice($stack['keys'], 0, MAX_KEY_COUNT);
-            call(getConsoleDB(), $stack['regions'], $chunk);
+            $chunk = array_slice($stack['keys'], 0, CHUNK_MAX_KEYS);
+            array_splice($stack['keys'], 0, CHUNK_MAX_KEYS);
+            Console::info("[{$time}] Sending " . count($chunk) . " remains " . count($stack['keys']));
+            call($stack['regions'], $chunk);
             $chunk = [];
         });
         echo "Out region [" . App::getEnv('_APP_REGION', 'nyc1') . "] cache purging worker Started" . PHP_EOL;