Contractless-PHP-RPC/src/Transport/EndpointPoolTransport.php

<?php

declare(strict_types=1);

namespace Contractless\Rpc\Transport;

use Closure;
use Contractless\Rpc\Exception\TransportException;
use InvalidArgumentException;

/**
 * Randomly distributes one-shot RPC calls across named transports.
 *
 * Each contained StreamTransport still opens a new socket, performs one
 * authenticated request, receives one reply, and closes the socket. The pool
 * stores endpoints, not persistent connections.
 */
final class EndpointPoolTransport implements TransportInterface
{
    /** @var array<string, TransportInterface> */
    private array $transports;

    private ?string $lastSuccessfulEndpointId = null;

    /**
     * @param array<string, TransportInterface> $transports
     */
    public function __construct(array $transports, private readonly bool $randomize = true)
    {
        if ($transports === []) {
            throw new InvalidArgumentException('The RPC endpoint pool cannot be empty.');
        }

        foreach ($transports as $endpointId => $transport) {
            if (!is_string($endpointId) || trim($endpointId) === '') {
                throw new InvalidArgumentException(
                    'Every RPC endpoint must have a non-empty string identifier.',
                );
            }
            if (!$transport instanceof TransportInterface) {
                throw new InvalidArgumentException(
                    "RPC endpoint $endpointId does not implement TransportInterface.",
                );
            }
        }

        $this->transports = $transports;
    }

    public function exchange(
        string $handshake,
        string $request,
        string $uid,
        Closure $validateHandshake,
    ): string {
        $this->lastSuccessfulEndpointId = null;
        $endpointIds = array_keys($this->transports);
        if ($this->randomize && count($endpointIds) > 1) {
            shuffle($endpointIds);
        }

        $failures = [];
        $lastFailure = null;

        foreach ($endpointIds as $endpointId) {
            try {
                $reply = $this->transports[$endpointId]->exchange(
                    $handshake,
                    $request,
                    $uid,
                    $validateHandshake,
                );
                $this->lastSuccessfulEndpointId = $endpointId;
                return $reply;
            } catch (TransportException $error) {
                // Only transport failures move the same request to another
                // endpoint. Protocol and validation errors remain final.
                $failures[] = "$endpointId: {$error->getMessage()}";
                $lastFailure = $error;
            }
        }

        throw new TransportException(
            'Every Contractless RPC endpoint failed: ' . implode('; ', $failures),
            previous: $lastFailure,
        );
    }

    /**
     * Return a specific transport for a linked request that must use the same
     * endpoint as an earlier operation, such as a paid storage lookup.
     */
    public function transportFor(string $endpointId): TransportInterface
    {
        if (!isset($this->transports[$endpointId])) {
            throw new InvalidArgumentException("Unknown RPC endpoint: $endpointId");
        }
        return $this->transports[$endpointId];
    }

    /** @return list<string> */
    public function endpointIds(): array
    {
        return array_keys($this->transports);
    }

    public function lastSuccessfulEndpointId(): ?string
    {
        return $this->lastSuccessfulEndpointId;
    }
}