"""Run a custom factory and borrowed stream against a local TCP echo endpoint.

Run ``uv run python examples/transports.py``. No external server is needed.
``tls_configuration`` shows a complete production-style TLS configuration;
the example only constructs it and does not read operator credential files.
"""

import argparse
import asyncio
import json
from importlib import import_module
from pathlib import Path

import anyio
from anyio.abc import ByteStream, SocketAttribute, SocketStream

from rpkiparrot.config import Endpoint, RtrSourceConfig
from rpkiparrot.transports import ConnectedTransport, TransportFactory


def tls_configuration(
    credential_directory: Path, *, default_trust: bool = False
) -> RtrSourceConfig:
    """Configure IP dialing with an independent DNS certificate identity.

    The operator issues router.pem with SAN iPAddress covering the source IP.
    RtrSession/Client context entry loads these files, and its TLS connection
    requires a verified server SAN dNSName for cache.example.net. Encrypted
    keys can instead be loaded into a host-created SSLContext with
    hostname_checks_common_name=False and injected as ssl_context.
    default_trust=True uses Python/OpenSSL CA paths, including the host's
    SSL_CERT_FILE / SSL_CERT_DIR settings. Otherwise only ca.pem is trusted.
    The trust profile labels that choice; it does not load any trust anchors.
    """
    return RtrSourceConfig(
        id="operator-cache",
        host="192.0.2.1",
        transport="tls",
        server_name="cache.example.net",
        local_address="192.0.2.2",
        trust_profile_id="host-default-ca-v1" if default_trust else "operator-ca-v1",
        ca_file=None if default_trust else credential_directory / "ca.pem",
        client_cert_file=credential_directory / "router.pem",
        client_key_file=credential_directory / "router.key",
    )


class LocalFactory:
    """An application-owned test transport; no reconnect property is required."""

    async def connect(self, endpoint: Endpoint) -> ByteStream:
        """Create a fresh stream whose ownership transfers to the caller."""
        assert endpoint.port is not None
        return await anyio.connect_tcp(endpoint.host, endpoint.port)


async def main(backend: str) -> None:
    """Exercise a factory, borrowed stream, and original-stream ownership."""
    try:
        asyncio.get_running_loop()
        actual_backend = "asyncio"
    except RuntimeError:
        # Optional runtime assertion; the asyncio example and its type check
        # remain usable in a core-only installation without Trio stubs.
        import_module("trio").lowlevel.current_task()
        actual_backend = "trio"
    assert actual_backend == backend
    config = tls_configuration(Path("credentials"))
    assert config.endpoint.server_name == "cache.example.net"
    default_config = tls_configuration(Path("credentials"), default_trust=True)
    assert default_config.ca_file is None
    assert default_config.client_cert_file == config.client_cert_file
    assert default_config.trust_profile_id != config.trust_profile_id

    async def echo(stream: SocketStream) -> None:
        async with stream:
            try:
                while True:
                    await stream.send(await stream.receive())
            except anyio.EndOfStream:
                return

    async with (
        await anyio.create_tcp_listener(local_host="127.0.0.1") as listener,
        anyio.create_task_group() as group,
    ):
        group.start_soon(listener.serve, echo)
        endpoint = Endpoint(
            transport="tcp", host="127.0.0.1", port=listener.extra(SocketAttribute.local_port)
        )
        factory: TransportFactory = LocalFactory()
        async with await factory.connect(endpoint) as original:
            borrowed = ConnectedTransport(original, owns_stream=False)
            async with await borrowed.connect(endpoint) as stream:
                await stream.send(b"borrowed stream")
                assert await stream.receive() == b"borrowed stream"
            await borrowed.aclose()
            await original.send(b"host still owns the connection")
            assert await original.receive() == b"host still owns the connection"
        group.cancel_scope.cancel()
    print(json.dumps({"backend": actual_backend, "borrowed_stream": "passed"}))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--backend", choices=("asyncio", "trio"), default="asyncio")
    selected_backend = parser.parse_args().backend
    anyio.run(main, selected_backend, backend=selected_backend)