<?php

declare (strict_types=1);
namespace GDAVSCOPE\Amp\Parallel\Ipc;

use GDAVSCOPE\Amp\Cache\LocalCache;
use GDAVSCOPE\Amp\Cancellation;
use GDAVSCOPE\Amp\DeferredFuture;
use GDAVSCOPE\Amp\ForbidCloning;
use GDAVSCOPE\Amp\ForbidSerialization;
use GDAVSCOPE\Amp\NullCancellation;
use GDAVSCOPE\Amp\Socket;
use GDAVSCOPE\Amp\Socket\ResourceSocket;
use GDAVSCOPE\Amp\Socket\SocketAddressType;
use GDAVSCOPE\Amp\TimeoutCancellation;
use GDAVSCOPE\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();
    }
    public function isClosed(): bool
    {
        return $this->server->isClosed();
    }
    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);
        }
    }
    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 ($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;
    }
}
