{

// 0.8.6 is last version with object-based headers

public const RESPONSE_FORMAT_OBJECT_HEADERS = '0.10.0';

// 0.9.0 is first version with array-based headers

public const RESPONSE_FORMAT_ARRAY_HEADERS = '0.11.0';

public const METHOD_GET = 'GET';

public const METHOD_POST = 'POST';

public const METHOD_DELETE = 'DELETE';

protected bool $selfSigned = false;

protected string $endpoint;

protected array $headers;

public function __construct()

{

$this->endpoint = System::getEnv('_APP_EXECUTOR_HOST', '');

$this->headers = [

'content-type' => 'application/json',

'authorization' => 'Bearer ' . System::getEnv('_APP_EXECUTOR_SECRET', ''),

'x-opr-addressing-method' => 'anycast-efficient',

'x-edge-bypass-gateway' => '1'

];

}

/**

public function deleteRuntime(string $projectId, string $deploymentId, string $suffix = '')

{

$runtimeId = "$projectId-$deploymentId" . $suffix;

$route = "/runtimes/$runtimeId";

$response = $this->call($this->endpoint, self::METHOD_DELETE, $route, [

'x-opr-addressing-method' => 'broadcast'

], [], true, 30);

$status = $response['headers']['status-code'];

$message = \is_string($response['body']) ? $response['body'] : ($response['body']['message'] ?? '');

// Runtime already gone — nothing to do

if ($status === 404) {

return true;

}

// Temporary fix for race condition

if ($status === 500 && \str_contains($message, 'already in progress')) {

return true; // OK, removal already in progress

}

if ($status >= 400) {

$type = \is_array($response['body']) ? ($response['body']['type'] ?? ExecutorException::GENERAL_UNKNOWN) : ExecutorException::GENERAL_UNKNOWN;

throw new ExecutorException($message, $status, type: $type);

}

return $response['body'];

}

/**

public function createExecution(

string $projectId,

string $deploymentId,

?string $body,

array $variables,

int $timeout,

string $image,

string $source,

string $entrypoint,

string $version,

string $path,

string $method,

array $headers,

float $cpus,

int $memory,

bool $logging,

string $runtimeEntrypoint = '',

?int $requestTimeout = null,

string $responseFormat = self::RESPONSE_FORMAT_OBJECT_HEADERS

) {

$runtimeId = "$projectId-$deploymentId";

$route = '/runtimes/' . $runtimeId . '/executions';

// Remove after migration

if ($version === 'v3' || $version === 'v4') {

$version = 'v5';

}

$params = [

'runtimeId' => $runtimeId,

'variables' => $variables,

'timeout' => $timeout,

'path' => $path,

'method' => $method,

'headers' => $headers,

'image' => $image,

'source' => $source,

'entrypoint' => $entrypoint,

'cpus' => $cpus,

'memory' => $memory,

'version' => $version,

'runtimeEntrypoint' => $runtimeEntrypoint,

'logging' => $logging,

'restartPolicy' => 'always' // Once utopia/orchestration has it, use DockerAPI::ALWAYS (0.13+)

];

if (!empty($body)) {

$params['body'] = $body;

}

// Safety timeout. Executor has timeout, and open runtime has soft timeout.

// This one shouldn't really happen, but prevents from unexpected networking behaviours.

if ($requestTimeout == null) {

$requestTimeout = $timeout + 15;

}

$response = $this->call($this->endpoint, self::METHOD_POST, $route, [ 'x-opr-runtime-id' => $runtimeId, 'content-type' => 'multipart/form-data', 'accept' => 'multipart/form-data', 'x-executor-response-format' => $responseFormat ], $params, true, $requestTimeout);

$status = $response['headers']['status-code'];

if ($status >= 400) {

$message = \is_string($response['body']) ? $response['body'] : ($response['body']['message'] ?? '');

$type = \is_array($response['body']) ? ($response['body']['type'] ?? ExecutorException::GENERAL_UNKNOWN) : ExecutorException::GENERAL_UNKNOWN;

throw new ExecutorException($message, $status, type: $type);

}

$headers = $response['body']['headers'] ?? [];

if (is_string($headers)) {

$headers = \json_decode($headers, true);

}

$response['body']['headers'] = $headers;

$response['body']['statusCode'] = \intval($response['body']['statusCode'] ?? 500);

$response['body']['duration'] = \floatval($response['body']['duration'] ?? 0);

$response['body']['startTime'] = \floatval($response['body']['startTime'] ?? \microtime(true));

return $response['body'];

}

/**

private function call(string $endpoint, string $method, string $path = '', array $headers = [], array $params = [], bool $decode = true, int $timeout = 15, ?callable $callback = null): array

{

$headers = array_merge($this->headers, $headers);

$ch = curl_init($endpoint . $path . (($method == self::METHOD_GET && !empty($params)) ? '?' . http_build_query($params) : ''));

$responseHeaders = [];

$responseStatus = -1;

$responseType = '';

$responseBody = '';

switch ($headers['content-type']) {

case 'application/json':

$query = json_encode($params);

break;

case 'multipart/form-data':

$multipart = new BodyMultipart();

foreach ($params as $key => $value) {

$multipart->setPart($key, $value);

}

$headers['content-type'] = $multipart->exportHeader();

$query = $multipart->exportBody();

break;

default:

$query = http_build_query($params);

break;

}

foreach ($headers as $i => $header) {

$headers[] = $i . ':' . $header;

unset($headers[$i]);

}

if (isset($callback)) {

$headers[] = 'accept: text/event-stream';

$handleEvent = function ($ch, $data) use ($callback) {

$callback($data);

return \strlen($data);

};

curl_setopt($ch, CURLOPT_WRITEFUNCTION, $handleEvent);

} else {

curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);

}

curl_setopt($ch, CURLOPT_CUSTOMREQUEST, $method);

curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);

curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);

curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 0);

curl_setopt($ch, CURLOPT_TIMEOUT, $timeout);

curl_setopt($ch, CURLOPT_HEADERFUNCTION, function ($curl, $header) use (&$responseHeaders) {

$len = strlen($header);

$header = explode(':', $header, 2);

if (count($header) < 2) { // ignore invalid headers

return $len;

}

$responseHeaders[strtolower(trim($header[0]))] = trim($header[1]);

return $len;

});

if ($method != self::METHOD_GET) {

curl_setopt($ch, CURLOPT_POSTFIELDS, $query);

}

// Allow self signed certificates

if ($this->selfSigned) {

curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, false);

curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false);

}

$responseBody = curl_exec($ch);

if (isset($callback)) {

return [];

}

$responseType = $responseHeaders['content-type'] ?? '';

$responseStatus = curl_getinfo($ch, CURLINFO_HTTP_CODE);

$curlError = curl_errno($ch);

$curlErrorMessage = curl_error($ch);

if ($decode) {

$strpos = strpos($responseType, ';');

$strpos = \is_bool($strpos) ? \strlen($responseType) : $strpos;

switch (substr($responseType, 0, $strpos)) {

case 'multipart/form-data':

$boundary = \explode('boundary=', $responseHeaders['content-type'])[1] ?? '';

$multipartResponse = new BodyMultipart($boundary);

$multipartResponse->load(\is_bool($responseBody) ? '' : $responseBody);

$responseBody = $multipartResponse->getParts();

break;

case 'application/json':

$json = json_decode($responseBody, true);

if ($json === null) {

throw new ExecutorException('Failed to parse response: ' . $responseBody);

}

$responseBody = $json;

$json = null;

break;

}

}

if ($curlError) {

if ($curlError == CURLE_OPERATION_TIMEDOUT) {

throw new ExecutorTimeout('Executor request timed out after ' . $timeout . ' seconds');

}

throw new ExecutorException($curlErrorMessage . ' with status code ' . $responseStatus, $responseStatus);

}

$responseHeaders['status-code'] = $responseStatus;

return [

'headers' => $responseHeaders,

'body' => $responseBody

];

}

/**

public function parseCookie(string $cookie): array

{

$cookies = [];

parse_str(strtr($cookie, array('&' => '%26', '+' => '%2B', ';' => '&')), $cookies);

return $cookies;

}

/**

protected function flatten(array $data, string $prefix = ''): array

{

$output = [];

foreach ($data as $key => $value) {

$finalKey = $prefix ? "{$prefix}[{$key}]" : $key;

if (is_array($value)) {

$output += $this->flatten($value, $finalKey); // @todo: handle name collision here if needed

} else {

$output[$finalKey] = $value;

}

}

return $output;

}