<?php declare(strict_types=1);

namespace Amp\Parallel\Ipc;

use Amp\Cache\LocalCache;
use Amp\Cancellation;
use Amp\DeferredFuture;
use Amp\ForbidCloning;
use Amp\ForbidSerialization;
use Amp\NullCancellation;
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;

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

    /** @var array<string, DeferredFuture> */
    private array $waitingByKey = [];

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

    private bool $queued = false;

    /** @var LocalCache<ResourceSocket> */
    private LocalCache $clientsByKey;

    /**
     * @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(),
        };

        $this->clientsByKey = new LocalCache(1024, $keyReceiveTimeout);

        $queued = &$this->queued;
        $waitingByKey = &$this->waitingByKey;
        $clientsByKey = &$this->clientsByKey;
        $this->accept = static function () use (
            &$queued,
            &$waitingByKey,
            &$clientsByKey,
            $server,
            $keyReceiveTimeout,
            $keyLength,
        ): void {
            while ($waitingByKey) {
                $client = $server->accept();
                if (!$client) {
                    $queued = false;
                    $exception = new Socket\SocketException('IPC socket closed before the client connected');
                    foreach ($waitingByKey as $deferred) {
                        $deferred->error($exception);
                    }
                    return;
                }

                try {
                    $received = readKey($client, new TimeoutCancellation($keyReceiveTimeout), $keyLength);
                } catch (\Throwable) {
                    $client->close();
                    continue; // Ignore possible foreign connection attempt.
                }

                if (isset($waitingByKey[$received])) {
                    $waitingByKey[$received]->complete($client);
                    unset($waitingByKey[$received]);
                } else {
                    $clientsByKey->set($received, $client);
                }
            }

            $queued = false;
        };
    }

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

    #[\Override]
    public function isClosed(): bool
    {
        return $this->server->isClosed();
    }

    #[\Override]
    public function close(): void
    {
        $this->server->close();

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

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

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

    #[\Override]
    public function getUri(): string
    {
        return $this->uri;
    }

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

    /**
     * @param string $key A key generated by {@see generateKey()}.
     */
    #[\Override]
    public function accept(string $key, ?Cancellation $cancellation = null): ResourceSocket
    {
        if ($this->server->isClosed()) {
            throw new Socket\SocketException('The IPC server has been closed');
        }

        if (\strlen($key) !== $this->keyLength) {
            throw new \ValueError(\sprintf(
                "Key provided is of length %d, expected %d",
                \strlen($key),
                $this->keyLength,
            ));
        }

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

        $client = $this->clientsByKey->get($key);
        if ($client !== null) {
            $this->clientsByKey->delete($key);

            return $client;
        }

        if (!$this->queued) {
            EventLoop::queue($this->accept);
            $this->queued = true;
        }

        $cancellation ??= new NullCancellation();

        $this->waitingByKey[$key] = $deferred = new DeferredFuture();

        $waitingByKey = &$this->waitingByKey;
        $cancellationId = $cancellation->subscribe(static function () use (&$waitingByKey, $key): void {
            unset($waitingByKey[$key]);
        });

        try {
            $client = $deferred->getFuture()->await($cancellation);
        } finally {
            $cancellation->unsubscribe($cancellationId);
        }

        return $client;
    }
}
