<?php

declare(strict_types=1);

namespace PhpWA\Transport\WebSocket;

use PhpWA\Crypto\RandomSourceInterface;
use PhpWA\Crypto\SystemRandomSource;
use PhpWA\Exception\ConnectionClosedException;
use PhpWA\Exception\ProtocolException;
use PhpWA\Exception\TimeoutException;
use PhpWA\Exception\TransportException;
use PhpWA\Runtime\Deadline;
use PhpWA\Transport\DirectStreamConnector;
use PhpWA\Transport\StreamConnectorInterface;

final class WebSocketClient implements WebSocketInterface
{
    private const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
    private const KEEPALIVE_INTERVAL_SECONDS = 20.0;
    private const STATE_CLOSED = 'closed';
    private const STATE_OPEN = 'open';
    private const STATE_CLOSING = 'closing';

    /** @var WebSocketClientOptions */ private $options;
    /** @var StreamConnectorInterface */ private $connector;
    /** @var RandomSourceInterface */ private $random;
    /** @var resource|null */ private $stream;
    /** @var string */ private $state = self::STATE_CLOSED;
    /** @var float */ private $lastActivityAt = 0.0;

    public function __construct(
        ?WebSocketClientOptions $options = null,
        ?StreamConnectorInterface $connector = null,
        ?RandomSourceInterface $random = null
    ) {
        $this->options = $options ?? new WebSocketClientOptions();
        $this->connector = $connector ?? new DirectStreamConnector();
        $this->random = $random ?? new SystemRandomSource();
    }

    public function connect(?Deadline $deadline = null): void
    {
        if ($this->isOpen()) { return; }
        $this->closeStream();
        $deadline = $deadline ?? Deadline::after($this->options->connectTimeout);
        $endpoint = $this->parseEndpoint();
        $stream = null;
        try {
            $stream = $this->connector->connect($endpoint['host'], $endpoint['port'], $endpoint['secure'], $deadline);
            if (!is_resource($stream)) {
                throw new TransportException('Stream connector returned an invalid stream.');
            }
            $this->stream = $stream;
            $this->performHandshake($endpoint, $deadline);
            $this->state = self::STATE_OPEN;
            $this->lastActivityAt = microtime(true);
        } catch (\Throwable $error) {
            if (is_resource($stream)) { fclose($stream); }
            $this->stream = null;
            $this->state = self::STATE_CLOSED;
            throw $error;
        }
    }

    public function sendBinary(string $payload, Deadline $deadline): void
    {
        if (strlen($payload) > $this->options->maxMessageBytes) {
            throw new ProtocolException('WebSocket message exceeds configured limit.');
        }
        $this->sendFrame(0x2, $payload, $deadline);
    }

    public function receiveBinary(Deadline $deadline): string
    {
        $message = '';
        $fragmented = false;
        try {
            while (true) {
                $frame = $this->readFrame($deadline);
                $opcode = $frame['opcode'];
                if ($opcode === 0x8) {
                    $this->handlePeerClose($frame['payload'], $deadline);
                }
                if ($opcode === 0x9) {
                    $this->sendFrame(0xA, $frame['payload'], $deadline);
                    continue;
                }
                if ($opcode === 0xA) { continue; }
                if ($opcode === 0x1) {
                    throw new ProtocolException('Text WebSocket messages are not valid on the WhatsApp binary transport.');
                }
                if ($opcode === 0x2) {
                    if ($fragmented) {
                        throw new ProtocolException('Interleaved WebSocket data frame during fragmentation.');
                    }
                    $message = $frame['payload'];
                    if ($frame['fin']) { return $message; }
                    $fragmented = true;
                    continue;
                }
                if ($opcode === 0x0) {
                    if (!$fragmented) {
                        throw new ProtocolException('Unexpected WebSocket continuation frame.');
                    }
                    if (strlen($message) + strlen($frame['payload']) > $this->options->maxMessageBytes) {
                        throw new ProtocolException('Fragmented WebSocket message exceeds configured limit.');
                    }
                    $message .= $frame['payload'];
                    if ($frame['fin']) { return $message; }
                }
            }
        } catch (TimeoutException $error) {
            if ($this->isOpen() && microtime(true) - $this->lastActivityAt >= self::KEEPALIVE_INTERVAL_SECONDS) {
                try {
                    $this->sendFrame(0x9, '', Deadline::after(min(1.0, $this->options->connectTimeout)));
                } catch (\Throwable $keepaliveError) {
                    $this->abort();
                    throw $keepaliveError;
                }
            }
            throw $error;
        } catch (ProtocolException $error) {
            $this->abort();
            throw $error;
        }
    }

    public function close(int $code = 1000, string $reason = ''): void
    {
        if (!$this->validCloseCode($code)) {
            throw new ProtocolException('Invalid WebSocket close code.');
        }
        if (preg_match('//u', $reason) !== 1 || strlen($reason) > 123) {
            throw new ProtocolException('WebSocket close reason is invalid or too long.');
        }
        if ($this->isOpen()) {
            $this->state = self::STATE_CLOSING;
            try {
                $this->sendFrame(0x8, pack('n', $code) . $reason, Deadline::after(min(1.0, $this->options->connectTimeout)), true);
            } catch (\Throwable $ignored) {
            }
        }
        $this->closeStream();
    }

    public function isOpen(): bool
    {
        return $this->state === self::STATE_OPEN && is_resource($this->stream) && !feof($this->stream);
    }

    /** @return array{scheme:string,host:string,port:int,path:string,secure:bool} */
    private function parseEndpoint(): array
    {
        $parts = parse_url($this->options->url);
        if (!is_array($parts) || !is_string($parts['scheme'] ?? null) || !is_string($parts['host'] ?? null)) {
            throw new TransportException('Invalid WebSocket URL.');
        }
        if (isset($parts['user']) || isset($parts['pass']) || isset($parts['fragment'])) {
            throw new TransportException('WebSocket URL credentials and fragments are not allowed.');
        }
        $scheme = strtolower($parts['scheme']);
        if ($scheme !== 'wss' && !($scheme === 'ws' && $this->options->allowInsecure)) {
            throw new TransportException('Secure wss:// WebSocket is required.');
        }
        $host = $parts['host'];
        if ($host === '' || preg_match('/[\x00-\x20\x7f]/', $host)) {
            throw new TransportException('Invalid WebSocket host.');
        }
        $secure = $scheme === 'wss';
        $port = isset($parts['port']) ? (int) $parts['port'] : ($secure ? 443 : 80);
        if ($port < 1 || $port > 65535) {
            throw new TransportException('Invalid WebSocket port.');
        }
        $path = is_string($parts['path'] ?? null) && $parts['path'] !== '' ? $parts['path'] : '/';
        if (isset($parts['query']) && $parts['query'] !== '') { $path .= '?' . $parts['query']; }
        if ($path[0] !== '/' || preg_match('/[\x00-\x20\x7f]/', $path)) {
            throw new TransportException('Invalid WebSocket request target.');
        }
        return ['scheme' => $scheme, 'host' => $host, 'port' => $port, 'path' => $path, 'secure' => $secure];
    }

    /** @param array{scheme:string,host:string,port:int,path:string,secure:bool} $endpoint */
    private function performHandshake(array $endpoint, Deadline $deadline): void
    {
        $keyBytes = $this->random->bytes(16);
        if (strlen($keyBytes) !== 16) {
            throw new TransportException('Random source returned an invalid WebSocket key.');
        }
        $key = base64_encode($keyBytes);
        $defaultPort = $endpoint['secure'] ? 443 : 80;
        $host = filter_var($endpoint['host'], FILTER_VALIDATE_IP, FILTER_FLAG_IPV6) !== false
            ? '[' . $endpoint['host'] . ']'
            : $endpoint['host'];
        $hostHeader = $endpoint['port'] === $defaultPort ? $host : $host . ':' . $endpoint['port'];
        $request = implode("\r\n", [
            'GET ' . $endpoint['path'] . ' HTTP/1.1',
            'Host: ' . $hostHeader,
            'Upgrade: websocket',
            'Connection: Upgrade',
            'Sec-WebSocket-Key: ' . $key,
            'Sec-WebSocket-Version: 13',
            'Origin: ' . $this->options->origin,
            '',
            '',
        ]);
        $this->writeAll($request, $deadline);
        $this->validateHandshake($this->readHttpHeaders($deadline), $key);
    }

    private function validateHandshake(string $rawHeaders, string $key): void
    {
        $lines = explode("\r\n", substr($rawHeaders, 0, -4));
        $status = array_shift($lines);
        if (!is_string($status) || preg_match('/^HTTP\/1\.[01] 101(?: |$)/D', $status) !== 1) {
            throw new ProtocolException('Invalid WebSocket handshake status.');
        }
        $headers = [];
        foreach ($lines as $line) {
            if ($line === '') { continue; }
            if ($line[0] === ' ' || $line[0] === "\t") {
                throw new ProtocolException('Folded WebSocket response headers are not accepted.');
            }
            $separator = strpos($line, ':');
            if ($separator === false) { throw new ProtocolException('Malformed WebSocket response header.'); }
            $name = strtolower(substr($line, 0, $separator));
            if (preg_match('/^[!#$%&\'*+.^_`|~0-9a-z-]+$/D', $name) !== 1) {
                throw new ProtocolException('Malformed WebSocket response header name.');
            }
            $headers[$name][] = trim(substr($line, $separator + 1));
        }
        if (!$this->headerHasToken($headers['upgrade'] ?? [], 'websocket')) {
            throw new ProtocolException('WebSocket Upgrade response header is invalid.');
        }
        if (!$this->headerHasToken($headers['connection'] ?? [], 'upgrade')) {
            throw new ProtocolException('WebSocket Connection response header is invalid.');
        }
        if (count($headers['sec-websocket-accept'] ?? []) !== 1) {
            throw new ProtocolException('WebSocket accept header must occur exactly once.');
        }
        $expected = base64_encode(sha1($key . self::GUID, true));
        if (!hash_equals($expected, $headers['sec-websocket-accept'][0])) {
            throw new ProtocolException('WebSocket accept key mismatch.');
        }
        if (isset($headers['sec-websocket-extensions']) || isset($headers['sec-websocket-protocol'])) {
            throw new ProtocolException('Server selected an unrequested WebSocket extension or subprotocol.');
        }
    }

    /** @param list<string> $values */
    private function headerHasToken(array $values, string $token): bool
    {
        foreach ($values as $value) {
            foreach (explode(',', $value) as $part) {
                if (strcasecmp(trim($part), $token) === 0) { return true; }
            }
        }
        return false;
    }

    private function sendFrame(int $opcode, string $payload, Deadline $deadline, bool $allowClosing = false): void
    {
        if (!$allowClosing && !$this->isOpen()) { throw new ConnectionClosedException('WebSocket is not open.'); }
        if ($allowClosing && (!is_resource($this->stream) || feof($this->stream))) { return; }
        if (!in_array($opcode, [0x2, 0x8, 0x9, 0xA], true)) {
            throw new ProtocolException('Unsupported client WebSocket opcode.');
        }
        $length = strlen($payload);
        if ($opcode >= 0x8 && $length > 125) { throw new ProtocolException('WebSocket control payload exceeds 125 bytes.'); }
        if ($length > $this->options->maxMessageBytes) { throw new ProtocolException('WebSocket frame exceeds configured limit.'); }

        $header = chr(0x80 | $opcode);
        if ($length <= 125) {
            $header .= chr(0x80 | $length);
        } elseif ($length <= 65535) {
            $header .= chr(0x80 | 126) . pack('n', $length);
        } else {
            $header .= chr(0x80 | 127) . pack('N2', intdiv($length, 4294967296), $length % 4294967296);
        }
        $mask = $this->random->bytes(4);
        if (strlen($mask) !== 4) { throw new TransportException('Random source returned an invalid WebSocket mask.'); }
        $maskStream = substr(str_repeat($mask, intdiv($length + 3, 4)), 0, $length);
        $masked = $length === 0 ? '' : ($payload ^ $maskStream);
        $this->writeAll($header . $mask . $masked, $deadline);
        $this->lastActivityAt = microtime(true);
    }

    /** @return array{fin:bool,opcode:int,payload:string} */
    private function readFrame(Deadline $deadline): array
    {
        if (!$this->isOpen()) { throw new ConnectionClosedException('WebSocket is not open.'); }
        $head = $this->readExact(2, $deadline);
        $first = ord($head[0]);
        $second = ord($head[1]);
        $fin = ($first & 0x80) !== 0;
        if (($first & 0x70) !== 0) { throw new ProtocolException('WebSocket reserved bits are not allowed.'); }
        $opcode = $first & 0x0f;
        if (!in_array($opcode, [0x0, 0x1, 0x2, 0x8, 0x9, 0xA], true)) {
            throw new ProtocolException('Unsupported server WebSocket opcode.');
        }
        if (($second & 0x80) !== 0) { throw new ProtocolException('Server WebSocket frames must not be masked.'); }
        $lengthCode = $second & 0x7f;
        $length = $lengthCode;
        if ($lengthCode === 126) {
            $value = unpack('nlength', $this->readExact(2, $deadline));
            $length = (int) $value['length'];
            if ($length < 126) { throw new ProtocolException('Non-minimal WebSocket length encoding.'); }
        } elseif ($lengthCode === 127) {
            $value = unpack('Nhigh/Nlow', $this->readExact(8, $deadline));
            $high = (int) $value['high'];
            $low = (int) $value['low'];
            if (($high & 0x80000000) !== 0) { throw new ProtocolException('WebSocket length has forbidden high bit.'); }
            $length = $high * 4294967296 + $low;
            if ($length <= 65535) { throw new ProtocolException('Non-minimal WebSocket length encoding.'); }
        }
        if ($opcode >= 0x8 && (!$fin || $length > 125)) {
            throw new ProtocolException('Invalid fragmented or oversized WebSocket control frame.');
        }
        if ($length > $this->options->maxMessageBytes) {
            throw new ProtocolException('Server WebSocket frame exceeds configured limit.');
        }
        $payload = $length === 0 ? '' : $this->readExact($length, $deadline);
        $this->lastActivityAt = microtime(true);
        return ['fin' => $fin, 'opcode' => $opcode, 'payload' => $payload];
    }

    private function handlePeerClose(string $payload, Deadline $deadline): void
    {
        if (strlen($payload) === 1) { throw new ProtocolException('Malformed WebSocket close payload.'); }
        $code = null;
        $reason = '';
        if ($payload !== '') {
            $unpacked = unpack('ncode', substr($payload, 0, 2));
            $code = (int) $unpacked['code'];
            $reason = substr($payload, 2);
            if (!$this->validCloseCode($code) || preg_match('//u', $reason) !== 1) {
                throw new ProtocolException('Invalid WebSocket peer close code or reason.');
            }
        }
        $this->state = self::STATE_CLOSING;
        try { $this->sendFrame(0x8, $payload, $deadline, true); } catch (\Throwable $ignored) {}
        $this->closeStream();
        throw new ConnectionClosedException('WebSocket was closed by the peer.', $code, $reason);
    }

    private function validCloseCode(int $code): bool
    {
        if ($code >= 3000 && $code <= 4999) { return true; }
        return $code >= 1000 && $code <= 1014 && !in_array($code, [1004, 1005, 1006], true);
    }

    private function readHttpHeaders(Deadline $deadline): string
    {
        $headers = '';
        while (strpos($headers, "\r\n\r\n") === false) {
            if (strlen($headers) >= $this->options->maxHeaderBytes) {
                throw new ProtocolException('WebSocket response headers exceed configured limit.');
            }
            $headers .= $this->readExact(1, $deadline);
        }
        return $headers;
    }

    private function readExact(int $length, Deadline $deadline): string
    {
        if ($length < 0) { throw new ProtocolException('Negative WebSocket read length.'); }
        if ($length === 0) { return ''; }
        $stream = $this->requireStream();
        $data = '';
        while (strlen($data) < $length) {
            $remaining = $deadline->remaining('WebSocket read');
            $this->setStreamTimeout($stream, $remaining);
            $chunk = fread($stream, $length - strlen($data));
            if ($chunk === false) {
                // TLS streams may return false (rather than an empty string)
                // when the underlying stream timeout expires. Preserve the
                // bounded-worker timeout contract instead of reporting a
                // misleading permanent transport failure.
                $metadata = stream_get_meta_data($stream);
                if (!empty($metadata['timed_out'])) {
                    throw new TimeoutException('Timed out during WebSocket read.');
                }
                if ($deadline->isExpired()) {
                    $deadline->remaining('WebSocket read');
                }
                if (feof($stream)) {
                    $this->abort();
                    throw new ConnectionClosedException('WebSocket stream ended unexpectedly.');
                }
                throw new TransportException('WebSocket read failed.');
            }
            if ($chunk === '') {
                $metadata = stream_get_meta_data($stream);
                if (!empty($metadata['timed_out'])) { throw new TimeoutException('Timed out during WebSocket read.'); }
                if (feof($stream)) {
                    $this->abort();
                    throw new ConnectionClosedException('WebSocket stream ended unexpectedly.');
                }
                usleep(1000);
                continue;
            }
            $data .= $chunk;
        }
        return $data;
    }

    private function writeAll(string $data, Deadline $deadline): void
    {
        $stream = $this->requireStream();
        $offset = 0;
        $length = strlen($data);
        while ($offset < $length) {
            $remaining = $deadline->remaining('WebSocket write');
            $this->setStreamTimeout($stream, $remaining);
            $written = fwrite($stream, substr($data, $offset));
            if ($written === false || $written === 0) {
                $metadata = stream_get_meta_data($stream);
                if (!empty($metadata['timed_out'])) { throw new TimeoutException('Timed out during WebSocket write.'); }
                throw new TransportException('WebSocket write failed.');
            }
            $offset += $written;
        }
    }

    /** @param resource $stream */
    private function setStreamTimeout($stream, float $seconds): void
    {
        $whole = (int) floor($seconds);
        $micros = (int) max(0, min(999999, ($seconds - $whole) * 1000000));
        stream_set_timeout($stream, $whole, $micros);
    }

    /** @return resource */
    private function requireStream()
    {
        if (!is_resource($this->stream)) { throw new ConnectionClosedException('WebSocket stream is unavailable.'); }
        return $this->stream;
    }

    private function abort(): void { $this->closeStream(); }

    private function closeStream(): void
    {
        if (is_resource($this->stream)) { fclose($this->stream); }
        $this->stream = null;
        $this->state = self::STATE_CLOSED;
        $this->lastActivityAt = 0.0;
    }
}
