App::getEnv('_APP_REGION') !== $region && $region !== 'default', ARRAY_FILTER_USE_KEY ); $stack = [ 'regions' => $regions, 'keys' => [], ]; $failures = []; const CHUNK_MAX_KEYS = 500; const MAX_CURL_SEND_ATTEMPTS = 4; /** * @param string $url * @param string $token * @param array data * @return int */ function call(string $url, string $token, array $data): int { $boundary = uniqid(); $delimiter = '-------------' . $boundary; $payload = ''; $eol = "\r\n"; console::warning('Sending ' . count($data) . 'to ' . $url); var_dump($data); foreach ($data as $keys) { $payload .= "--" . $delimiter . $eol . 'Content-Disposition: form-data; name="keys[]"' . $eol . $eol . json_encode($keys) . $eol; } $payload .= "--" . $delimiter . "--" . $eol; $status = 404; $ch = curl_init($url); curl_setopt($ch, CURLOPT_HTTPHEADER, [ 'Authorization: Bearer ' . $token, 'Origin-edge-url: ' . App::getEnv('_APP_REGION'), 'Content-type: multipart/form-data; boundary=' . $delimiter, 'Content-Length: ' . strlen($payload) ]); curl_setopt($ch, CURLOPT_RETURNTRANSFER, true); curl_setopt($ch, CURLOPT_TIMEOUT, 5); curl_setopt($ch, CURLOPT_POST, true); curl_setopt($ch, CURLOPT_POSTFIELDS, $payload); curl_setopt($ch, CURLOPT_VERBOSE, true); curl_setopt($ch, CURLOPT_HEADER, false); curl_setopt($ch, CURLOPT_CUSTOMREQUEST, "POST"); for ($attempts = 0; $attempts < MAX_CURL_SEND_ATTEMPTS; $attempts++) { curl_exec($ch); $status = curl_getinfo($ch, CURLINFO_HTTP_CODE); if ($status === 200) { return $status; } sleep(1); } curl_close($ch); return $status; } /** * @throws Authorization * @throws Structure * @throws Exception|\Exception */ function handle($dbForConsole, $regions, $data): void { $jwt = new JWT(App::getEnv('_APP_OPENSSL_KEY_V1'), 'HS256', 600, 10); $token = $jwt->encode([]); foreach ($regions as $code => $region) { $time = DateTime::now(); $status = call($region['domain'] . '/v1/edge/sync', $token, $data); if ($status !== Response::STATUS_CODE_OK) { Console::error("[{$time}] Request to {$code} has failed"); foreach ($data as $keys) { $dbForConsole->createDocument('syncs', new Document([ 'region' => App::getEnv('_APP_REGION'), 'target' => $code, 'type' => $keys['type'], 'key' => ['key' => $keys['key']], 'status' => $status, ])); } } } } $server->job() ->inject('message') ->action(function (Message $message) use (&$stack, &$failures) { $payload = $message->getPayload() ?? []; //Get failed requests if (!empty($payload['region']) && !empty($payload['keys'])) { $regions = array_filter( Config::getParam('regions', []), fn ($region) => $payload['region'] === $region, ARRAY_FILTER_USE_KEY ); $failures[] = [ 'regions' => $regions, 'keys' => $payload['keys'] ]; return; } if (empty($payload['type'])) { return; } if (!empty($payload['key'])) { $stack['keys'][] = [ 'type' => $payload['type'], 'key' => $payload['key'], ]; } }); $server ->workerStart() ->inject('dbForConsole') ->action(function ($dbForConsole) use (&$stack, &$failures) { Timer::tick(5000, function () use ($dbForConsole, &$stack, &$failures) { $time = DateTime::now(); if (empty($stack['keys']) && count($failures) === 0) { Console::info("[{$time}] Stack is empty"); return; } //Send failed requests if (count($failures) > 0) { $i = 0; while ($i < count($failures)) { $failure = array_shift($failures); Console::info("[{$time}] ReSending " . count($failure['keys']) . " to " . key($failure['regions'])); handle($dbForConsole, $failure['regions'], $failure['keys']); $i++; } return; } $chunk = array_slice($stack['keys'], 0, CHUNK_MAX_KEYS, true); array_splice($stack['keys'], 0, CHUNK_MAX_KEYS); Console::log("[{$time}] Sending " . count($chunk) . " remains " . count($stack['keys'])); handle($dbForConsole, $stack['regions'], $chunk); }); Console::success("[" . App::getEnv('_APP_REGION') . "] edge sync-out worker Started"); }); $server->start();