<?php declare(strict_types=1);

namespace Amp\Parallel\Ipc;

use Amp\Cancellation;
use Amp\DeferredFuture;
use Amp\ForbidCloning;
use Amp\ForbidSerialization;
use Amp\Socket;
use Amp\Socket\ResourceSocket;
use Amp\Socket\SocketAddressType;
use Amp\TimeoutCancellation;
use Revolt\EventLoop;

final class SocketIpcHub implements IpcHub
{
    use ForbidCloning;
    use ForbidSerialization;

    public const DEFAULT_KEY_RECEIVE_TIMEOUT = 5;
    public const DEFAULT_KEY_LENGTH = 64;

    private int $nextId = 0;

    /** @var non-empty-string */
    private readonly string $uri;

    /** @var array<string, int> */
    private array $keys = [];

    /** @var array<int, DeferredFuture> */
    private array $pending = [];

    /** @var \Closure(): void */
    private readonly \Closure $accept;

    /**
     * @param float $keyReceiveTimeout Timeout to receive the key on accepted connections.
     * @param positive-int $keyLength Length of the random key exchanged on the IPC channel when connecting.
     */
    public function __construct(
        private readonly Socket\ServerSocket $server,
        float $keyReceiveTimeout = self::DEFAULT_KEY_RECEIVE_TIMEOUT,
        private readonly int $keyLength = self::DEFAULT_KEY_LENGTH,
    ) {
        $address = $this->server->getAddress();
        $this->uri = match ($address->getType()) {
            SocketAddressType::Unix => 'unix://' . $address->toString(),
            SocketAddressType::Internet => 'tcp://' . $address->toString(),
        };

        $keys = &$this->keys;
        $pending = &$this->pending;
        $this->accept = static function () use (&$keys, &$pending, $server, $keyReceiveTimeout, $keyLength): void {
            while ($pending && $client = $server->accept()) {
                try {
                    $received = readKey($client, new TimeoutCancellation($keyReceiveTimeout), $keyLength);
                } catch (\Throwable) {
                    $client->close();
                    continue; // Ignore possible foreign connection attempt.
                }

                $id = $keys[$received] ?? null;

                if ($id === null) {
                    $client->close();
                    continue; // Ignore possible foreign connection attempt.
                }

                $deferred = $pending[$id] ?? null;
                unset($pending[$id], $keys[$received]);

                if ($deferred === null) {
                    $client->close();
                    continue; // Client accept cancelled.
                }

                $deferred->complete($client);
            }
        };
    }

    public function __destruct()
    {
        $this->close();
    }

    public function isClosed(): bool
    {
        return $this->server->isClosed();
    }

    public function close(): void
    {
        $this->server->close();

        if (!$this->pending) {
            return;
        }

        $exception = new Socket\SocketException('IPC socket closed before the client connected');
        foreach ($this->pending as $deferred) {
            $deferred->error($exception);
        }
    }

    public function onClose(\Closure $onClose): void
    {
        $this->server->onClose($onClose);
    }

    public function getUri(): string
    {
        return $this->uri;
    }

    public function generateKey(): string
    {
        return \random_bytes($this->keyLength);
    }

    /**
     * @param string $key A key generated by {@see generateKey()}.
     */
    public function accept(string $key, ?Cancellation $cancellation = null): ResourceSocket
    {
        if (\strlen($key) !== $this->keyLength) {
            throw new \ValueError(\sprintf(
                "Key provided is of length %d, expected %d",
                \strlen($key),
                $this->keyLength,
            ));
        }

        if (isset($this->keys[$key])) {
            throw new \Error("An accept is already pending for the given key");
        }

        $id = $this->nextId++;

        if (!$this->pending) {
            EventLoop::queue($this->accept);
        }

        $this->keys[$key] = $id;
        $this->pending[$id] = $deferred = new DeferredFuture;

        try {
            $client = $deferred->getFuture()->await($cancellation);
        } finally {
            unset($this->pending[$id], $this->keys[$key]);
        }

        return $client;
    }
}
