<?php

namespace App\Xapi;

use JumPHP\Async\Task;
use JumPHP\Kernel\Platform;

/**
 * One big HTTP transfer on a stream of its own (a VM exported as an XVA, an XVA imported): XAPI's HTTP handlers GET /export and
 * PUT /import. Never the whole file in memory, never a blocking wait:
 *  - download(): the body goes to the file chunk by chunk as it arrives (Content-Length, chunked, or until the server closes);
 *  - upload(): the file is read and written to the stream a chunk at a time, at most $budget bytes a frame, and not while the
 *    stream says it still holds more than HIGH_WATER bytes to send (backpressure: TlsStream::buffered()).
 * progress() = bytes / Content-Length (or the size expected). The Task settles with the HttpReply of the server (its body: the
 * small text of an answer, never the file); a failure of the network fails it.
 *
 *   $t = HttpStreamTransfer::download($connect, $host, '/export?session_id=…&ref=…&use_compression=true', '/home/me/web-01.xva');
 *   $t->task()->then(fn (HttpReply $r) => ..., fn (\Throwable $e) => ...);      // and $t->poll($dt) each frame
 */
final class HttpStreamTransfer
{
    public const CHUNK = 65536;
    public const HIGH_WATER = 1048576;
    /** seconds without a byte in either direction: the transfer failed */
    public const IDLE = 120.0;

    public string $state = 'connecting';
    /** bytes of the body written to the file (download) or sent (upload) */
    public float $bytes = 0.0;
    /** the size of the body: Content-Length, the file's size, or what the caller expects (0: unknown) */
    public float $total = 0.0;
    /** bytes written to the stream per frame at most (upload) */
    public float $budget;
    public int $status = 0;
    /** @var (\Closure(self): void)|null called each frame while it runs (the progress of an Events line) */
    public ?\Closure $onProgress = null;
    private Task $task;
    private ?object $stream = null;
    private bool $open = false;
    private string $head = '';
    private bool $headDone = false;
    /** @var array<string, string> */
    private array $headers = [];
    private string $replyBody = '';
    /** @var resource|null */
    private $file = null;
    private float $idle = 0.0;
    // chunked decoding, as the bytes come
    private bool $chunked = false;
    private float $chunkLeft = -1;
    private string $chunkLine = '';
    private bool $saving = false;
    private bool $finished = false;

    /** @param \Closure(): object $connect */
    private function __construct(private \Closure $connect, private string $hostHeader, private string $method, private string $path, private string $local)
    {
        $this->task = new Task();
        $this->budget = Platform::isWasm() ? 262144.0 : 1048576.0;
    }

    /** GET $path, its body into $file (created when the server answers 200; removed if the transfer fails). */
    public static function download(\Closure $connect, string $hostHeader, string $path, string $file, float $expected = 0.0): self
    {
        $t = new self($connect, $hostHeader, 'GET', $path, $file);
        $t->total = $expected;
        $t->start();
        return $t;
    }

    /** PUT $path with $file as the body (Content-Length: its size). */
    public static function upload(\Closure $connect, string $hostHeader, string $path, string $file, string $method = 'PUT'): self
    {
        $t = new self($connect, $hostHeader, $method, $path, $file);
        $size = @filesize($file);
        $handle = $size === false ? false : @fopen($file, 'rb');
        if ($handle === false) {
            $t->fail("Cannot read the file $file.");
            return $t;
        }
        $t->file = $handle;
        $t->total = (float)$size;
        $t->start();
        return $t;
    }

    public function task(): Task
    {
        return $this->task;
    }

    public function isDone(): bool
    {
        return $this->finished;
    }

    public function progress(): float
    {
        return $this->total > 0 ? max(0.0, min(1.0, $this->bytes / $this->total)) : 0.0;
    }

    private function start(): void
    {
        try {
            $this->stream = ($this->connect)();
        } catch (\Throwable $e) {
            $this->fail($e->getMessage());
            return;
        }
        $this->stream->on('open', function (): void {
            if ($this->finished) {
                return;
            }
            $this->open = true;
            $this->state = $this->method === 'GET' ? 'receiving' : 'sending';
            $lines = "{$this->method} {$this->path} HTTP/1.1\r\nHost: {$this->hostHeader}\r\nUser-Agent: XenCenter-JumPHP\r\nConnection: close\r\n";
            if ($this->method !== 'GET') {
                $lines .= 'Content-Length: ' . sprintf('%.0F', $this->total) . "\r\nContent-Type: application/octet-stream\r\n";
            }
            $this->stream?->write($lines . "\r\n");
        });
        $this->stream->on('data', fn (string $bytes) => $this->received($bytes));
        $this->stream->on('error', fn (string $message) => $this->fail($message));
        $this->stream->on('close', function (?string $why = null): void {
            if ($this->finished) {
                return;
            }
            if ($this->headDone && !$this->chunked && !isset($this->headers['content-length'])) {
                $this->complete();          // a body read until the server closes
                return;
            }
            $this->fail($why ?? ($this->headDone ? 'The connection closed before the end of the transfer.' : 'The server closed the connection without answering.'));
        });
    }

    /** Each frame: the next chunks of an upload, the idle timeout. */
    public function poll(float $dt): void
    {
        if ($this->finished) {
            return;
        }
        $this->idle += $dt;
        if ($this->onProgress !== null) {
            ($this->onProgress)($this);
        }
        if ($this->idle > self::IDLE) {
            $this->fail('No data for ' . (int)self::IDLE . ' s: the transfer stopped.');
            return;
        }
        if ($this->state !== 'sending' || !$this->open || $this->file === null || $this->stream === null) {
            return;
        }
        $sent = 0.0;
        while ($sent < $this->budget && $this->bytes < $this->total) {
            if (method_exists($this->stream, 'buffered') && $this->stream->buffered() > self::HIGH_WATER) {
                return;                     // the stream still holds a lot to send: wait for it
            }
            $chunk = fread($this->file, (int)min(self::CHUNK, $this->total - $this->bytes));
            if ($chunk === false || $chunk === '') {
                $this->fail('The file ended before its size (' . sprintf('%.0F', $this->total) . ' bytes): it changed while it was sent.');
                return;
            }
            if ($this->stream->write($chunk) === false) {
                $this->fail('The connection closed while the file was sent.');
                return;
            }
            $this->bytes += strlen($chunk);
            $sent += strlen($chunk);
            $this->idle = 0;
        }
        if ($this->bytes >= $this->total) {
            fclose($this->file);
            $this->file = null;
            $this->state = 'waiting';       // the whole body is out: the server's answer comes next
        }
    }

    public function cancel(string $why = 'Cancelled.'): void
    {
        $this->fail($why);
    }

    private function received(string $bytes): void
    {
        if ($this->finished) {
            return;
        }
        $this->idle = 0;
        if (!$this->headDone) {
            $this->head .= $bytes;
            $end = strpos($this->head, "\r\n\r\n");
            if ($end === false) {
                if (strlen($this->head) > 65536) {
                    $this->fail('The answer of the server is not HTTP.');
                }
                return;
            }
            $rest = (string)substr($this->head, $end + 4);
            $this->parseHead(substr($this->head, 0, $end));
            $this->head = '';
            $this->headDone = true;
            if ($this->method !== 'GET' && $this->state === 'sending' && $this->status >= 300) {
                // the server refused the upload before its end (no session, no space...): stop sending
                $this->closeFile();
                $this->state = 'waiting';
            }
            $this->saving = $this->method === 'GET' && $this->status === 200;
            if ($this->saving) {
                $handle = @fopen($this->local, 'wb');
                if ($handle === false) {
                    $this->fail("Cannot write the file {$this->local}.");
                    return;
                }
                $this->file = $handle;
                if (isset($this->headers['content-length'])) {
                    $this->total = (float)$this->headers['content-length'];
                }
            }
            if ($rest === '' && isset($this->headers['content-length']) && (float)$this->headers['content-length'] == 0) {
                $this->complete();
                return;
            }
            $bytes = $rest;
            if ($bytes === '') {
                return;
            }
        }
        $this->chunked ? $this->dechunk($bytes) : $this->body($bytes);
        if (!$this->finished && !$this->chunked && isset($this->headers['content-length']) && $this->bodyBytes() >= (float)$this->headers['content-length']) {
            $this->complete();
        }
    }

    private function bodyBytes(): float
    {
        return $this->saving ? $this->bytes : (float)strlen($this->replyBody);
    }

    /** Bytes of the body: to the file (a download that answered 200), else kept as the answer's text (small). */
    private function body(string $bytes): void
    {
        if ($this->saving && $this->file !== null) {
            if (@fwrite($this->file, $bytes) !== strlen($bytes)) {
                $this->fail("Cannot write the file {$this->local} (disk full?).");
                return;
            }
            $this->bytes += strlen($bytes);
        } elseif (strlen($this->replyBody) < 65536) {
            $this->replyBody .= $bytes;
        }
    }

    /** Transfer-Encoding: chunked, decoded as it comes: "<hex size>\r\n<bytes>\r\n" ... "0\r\n\r\n". */
    private function dechunk(string $data): void
    {
        while ($data !== '' && !$this->finished) {
            if ($this->chunkLeft > 0) {
                $part = (string)substr($data, 0, (int)min($this->chunkLeft, strlen($data)));
                $this->body($part);
                $this->chunkLeft -= strlen($part);
                $data = (string)substr($data, strlen($part));
                if ($this->chunkLeft == 0) {
                    $this->chunkLeft = -2;      // its CRLF comes next
                }
                continue;
            }
            $this->chunkLine .= $data;
            $data = '';
            if ($this->chunkLeft == -2) {
                if (strlen($this->chunkLine) < 2) {
                    return;
                }
                $this->chunkLine = (string)substr($this->chunkLine, 2);
                $this->chunkLeft = -1;
            }
            $eol = strpos($this->chunkLine, "\r\n");
            if ($eol === false) {
                return;
            }
            $size = (float)hexdec(trim(explode(';', substr($this->chunkLine, 0, $eol))[0]));
            $data = (string)substr($this->chunkLine, $eol + 2);
            $this->chunkLine = '';
            if ($size == 0) {
                $this->complete();              // the trailers do not matter
                return;
            }
            $this->chunkLeft = $size;
        }
    }

    private function parseHead(string $head): void
    {
        $lines = explode("\r\n", $head);
        $this->status = preg_match('#^HTTP/\d(?:\.\d)?\s+(\d{3})#', (string)array_shift($lines), $m) ? (int)$m[1] : 0;
        foreach ($lines as $line) {
            $colon = strpos($line, ':');
            if ($colon !== false) {
                $this->headers[strtolower(trim(substr($line, 0, $colon)))] = trim(substr($line, $colon + 1));
            }
        }
        $this->chunked = str_contains(strtolower($this->headers['transfer-encoding'] ?? ''), 'chunked');
    }

    private function complete(): void
    {
        if ($this->finished) {
            return;
        }
        $this->finished = true;
        $this->state = 'done';
        $this->closeFile();
        $stream = $this->stream;
        $this->stream = null;
        $stream?->close();
        $this->task->settle(new HttpReply($this->status, $this->headers, $this->replyBody));
    }

    private function fail(string $message): void
    {
        if ($this->finished) {
            return;
        }
        $this->finished = true;
        $this->state = 'failed';
        $this->closeFile();
        if ($this->saving) {
            @unlink($this->local);              // a part of an XVA is no XVA
        }
        $stream = $this->stream;
        $this->stream = null;
        $stream?->close();
        $this->task->settle(null, XapiError::transport($message));
    }

    private function closeFile(): void
    {
        if ($this->file !== null) {
            fclose($this->file);
            $this->file = null;
        }
    }
}