<?php

namespace App\Xapi;

use JumPHP\Async\Task;

/**
 * HTTP/1.1 over one stream (a TcpConnection, an SshConnection, a TlsStream: anything with on() / write() / close()), the
 * requests one after the other on the same connection (keep-alive). The stream is opened when the first request comes, and
 * again after the server closed it. Answers with Content-Length, chunked, or "read until close".
 */
final class HttpChannel
{
    /** @var list<array{0: string, 1: Task}> requests waiting (raw request, task) */
    private array $queue = [];
    /** @var array{0: string, 1: Task}|null */
    private ?array $current = null;
    private ?object $stream = null;
    private bool $open = false;
    private string $buffer = '';
    private float $waited = 0;
    /** @var int a counter: events of an old stream are ignored */
    private int $generation = 0;
    /** answers given on the current stream: a request written on a reused stream that dies before any byte is sent again, once */
    private int $served = 0;
    private bool $onReused = false;
    private bool $retried = false;
    /** @var array{0: int, 1: array<string, string>}|null the current answer's status and headers, once its head arrived */
    private ?array $head = null;
    private int $scanFrom = 0;
    private int $bodyAt = 0;
    private int $chunkAt = 0;
    private string $decoded = '';

    /** @param \Closure(): object $connect opens a new stream */
    public function __construct(private \Closure $connect, private string $hostHeader, private float $timeout = 40.0)
    {
    }

    /** @param array<string, string> $headers */
    public function request(string $method, string $path, string $body = '', array $headers = []): Task
    {
        $task = new Task();
        $lines = "$method $path HTTP/1.1\r\nHost: {$this->hostHeader}\r\nUser-Agent: XenCenter-JumPHP\r\nConnection: keep-alive\r\n";
        foreach ($headers as $name => $value) {
            $lines .= "$name: $value\r\n";
        }
        if ($body !== '' || $method === 'POST') {
            $lines .= 'Content-Length: ' . strlen($body) . "\r\n";
        }
        $this->queue[] = [$lines . "\r\n" . $body, $task];
        $this->next();
        return $task;
    }

    public function pending(): int
    {
        return count($this->queue) + ($this->current !== null ? 1 : 0);
    }

    public function poll(float $dt): void
    {
        if ($this->current === null) {
            return;
        }
        $this->waited += $dt;
        if ($this->waited > $this->timeout) {
            $this->failCurrent('The server did not answer in ' . (int)$this->timeout . ' s.');
            $this->drop();
            $this->next();
        }
    }

    public function close(): void
    {
        foreach ($this->queue as [, $task]) {
            $task->settle(null, XapiError::transport('Disconnected.'));
        }
        $this->queue = [];
        $this->failCurrent('Disconnected.');
        $this->drop();
    }

    private function next(): void
    {
        if ($this->current !== null || $this->queue === []) {
            return;
        }
        $this->current = array_shift($this->queue);
        $this->waited = 0;
        $this->buffer = '';
        $this->resetAnswer();
        $this->onReused = $this->stream !== null && $this->served > 0;
        if ($this->stream === null) {
            $this->openStream();
        }
        if ($this->open && $this->current !== null) {
            $this->stream->write($this->current[0]);
        }
    }

    private function openStream(): void
    {
        $generation = ++$this->generation;
        $this->open = false;
        $this->served = 0;
        try {
            $stream = ($this->connect)();
        } catch (\Throwable $e) {
            $this->failCurrent($e->getMessage());
            $this->next();              // the others queued behind it try in turn (each fails the same way, nothing waits forever)
            return;
        }
        $this->stream = $stream;
        $stream->on('open', function () use ($generation): void {
            if ($generation !== $this->generation) {
                return;
            }
            $this->open = true;
            if ($this->current !== null) {
                $this->stream?->write($this->current[0]);
            }
        });
        $stream->on('data', function (string $bytes) use ($generation): void {
            if ($generation === $this->generation) {
                $this->buffer .= $bytes;
                $this->waited = 0;          // the timeout counts silence, not the length of a big answer still arriving
                $this->parse(false);
            }
        });
        $stream->on('error', function (string $message) use ($generation): void {
            if ($generation === $this->generation) {
                $this->failCurrent($message);
                $this->drop();
                $this->retryLater();
            }
        });
        $stream->on('close', function (?string $why = null) use ($generation): void {
            if ($generation !== $this->generation) {
                return;
            }
            // an answer read until the end of the stream (parse() then goes on with the next request, on a new stream)
            if ($this->current !== null && $this->buffer !== '') {
                $this->parse(true);
                if ($generation !== $this->generation) {
                    return;
                }
            }
            if ($this->current !== null) {
                $this->failCurrent($why ?? 'The connection closed.');
            }
            $this->drop();
            $this->retryLater();
        });
    }

    private function retryLater(): void
    {
        // the next queued request opens a new stream (keep-alive ended, a failure): never in a loop, one at a time
        $this->next();
    }

    private function drop(): void
    {
        $this->generation++;
        $stream = $this->stream;
        $this->stream = null;
        $this->open = false;
        $stream?->close();
    }

    private function failCurrent(string $message): void
    {
        if ($this->current === null) {
            return;
        }
        if ($this->onReused && $this->buffer === '' && !$this->retried) {
            // a kept-alive connection the server (a NAT, the bridge) had closed while idle: the request never arrived, send it again
            array_unshift($this->queue, $this->current);
            $this->current = null;
            $this->onReused = false;
            $this->retried = true;
            return;
        }
        $this->retried = false;
        [, $task] = $this->current;
        $this->current = null;
        $task->settle(null, XapiError::transport($message));
    }

    /**
     * One answer in the buffer? $atEnd: the stream ended (an answer without length is complete). Incremental: the head is parsed
     * once, a body is only measured until it is whole, a chunked body is decoded as its chunks come. Each byte is copied once, not
     * once per piece received: the first event.from of a pool is tens of MB, in the browser in thousands of WebSocket messages
     * (re-reading the whole buffer each time took minutes there: stuck on "Synchronizing…").
     */
    private function parse(bool $atEnd): void
    {
        if ($this->current === null) {
            return;
        }
        if ($this->head === null) {
            $end = strpos($this->buffer, "\r\n\r\n", $this->scanFrom);
            if ($end === false) {
                $this->scanFrom = max(0, strlen($this->buffer) - 3);
                return;
            }
            $lines = explode("\r\n", substr($this->buffer, 0, $end));
            $status = preg_match('#^HTTP/\d(?:\.\d)?\s+(\d{3})#', (string)array_shift($lines), $m) ? (int)$m[1] : 0;
            $headers = [];
            foreach ($lines as $line) {
                $colon = strpos($line, ':');
                if ($colon !== false) {
                    $headers[strtolower(trim(substr($line, 0, $colon)))] = trim(substr($line, $colon + 1));
                }
            }
            $this->head = [$status, $headers];
            $this->bodyAt = $end + 4;
            $this->chunkAt = $this->bodyAt;
            $this->decoded = '';
        }
        [$status, $headers] = $this->head;
        $closeAfter = strtolower($headers['connection'] ?? '') === 'close';
        if (isset($headers['content-length'])) {
            $length = (int)$headers['content-length'];
            if (strlen($this->buffer) - $this->bodyAt < $length) {
                return;
            }
            $body = substr($this->buffer, $this->bodyAt, $length);
            $this->buffer = substr($this->buffer, $this->bodyAt + $length);
        } elseif (str_contains(strtolower($headers['transfer-encoding'] ?? ''), 'chunked')) {
            $rest = $this->dechunkMore();
            if ($rest === null) {
                return;
            }
            $body = $this->decoded;
            $this->buffer = $rest;
        } elseif ($status === 204 || $status === 304) {
            $body = '';
            $this->buffer = substr($this->buffer, $this->bodyAt);
        } else {
            if (!$atEnd) {
                return;             // read until the server closes
            }
            $body = substr($this->buffer, $this->bodyAt);
            $this->buffer = '';
            $closeAfter = true;
        }
        $this->resetAnswer();
        [, $task] = $this->current;
        $this->current = null;
        if ($closeAfter) {
            $this->drop();
        }
        $this->served++;
        $this->retried = false;
        $task->settle(new HttpReply($status, $headers, $body));
        $this->next();
    }

    /** The chunks complete so far go into $decoded; @return string|null what follows the last chunk, null while incomplete */
    private function dechunkMore(): ?string
    {
        $data = $this->buffer;
        $at = $this->chunkAt;
        while (true) {
            $eol = strpos($data, "\r\n", $at);
            if ($eol === false) {
                return null;
            }
            $size = (int)hexdec(trim(explode(';', substr($data, $at, $eol - $at))[0]));
            if ($size === 0) {
                $end = strpos($data, "\r\n", $eol + 2);       // trailers end with an empty line
                return $end === false ? null : (string)substr($data, $end + 2);
            }
            if (strlen($data) < $eol + 2 + $size + 2) {
                return null;
            }
            $this->decoded .= substr($data, $eol + 2, $size);
            $at = $eol + 2 + $size + 2;
            $this->chunkAt = $at;
        }
    }

    private function resetAnswer(): void
    {
        $this->head = null;
        $this->scanFrom = 0;
        $this->bodyAt = 0;
        $this->chunkAt = 0;
        $this->decoded = '';
    }

    /** @return array{0: string, 1: string}|null the body and what follows it, null while incomplete */
    public static function dechunk(string $data): ?array
    {
        $body = '';
        $at = 0;
        while (true) {
            $eol = strpos($data, "\r\n", $at);
            if ($eol === false) {
                return null;
            }
            $size = hexdec(trim(explode(';', substr($data, $at, $eol - $at))[0]));
            $at = $eol + 2;
            if ($size == 0) {
                $end = strpos($data, "\r\n", $at);       // trailers end with an empty line
                return $end === false ? null : [$body, (string)substr($data, $end + 2)];
            }
            if (strlen($data) < $at + $size + 2) {
                return null;
            }
            $body .= substr($data, $at, (int)$size);
            $at += (int)$size + 2;
        }
    }
}