"""Start a frozen meshbay-node as a connection process and carry bytes through it.

With `connection_processes` above 1 the node starts copies of itself as
`meshbay-node.exe webrtc-worker` (transport/webrtc/workers.py). The daemon's own
smoke test never gets there: the processes start once the node has signed in to
a hub, and that test has none. This is the node's side of it, unchanged, with a
frozen executable as the process and an aiortc peer here as the browser.

Run by build-node-runtime.ps1 with the build venv's interpreter, which has the
node's own modules. It takes any meshbay-node.exe, an installed one included
(`resources/node-runtime` under the application's directory).
"""

import asyncio
import sys
import time

from aiortc import RTCPeerConnection, RTCSessionDescription
from meshbay_node.transport.webrtc import workers

CHUNK = bytes(48_000)
CHUNKS = 200


async def until(test, what: str, seconds: float = 30.0) -> None:
    deadline = time.monotonic() + seconds
    while not test():
        if time.monotonic() > deadline:
            raise SystemExit(f"FAILED  {what}: not within {seconds:.0f}s")
        await asyncio.sleep(0.02)


async def carry(exe: str) -> str:
    workers._argv = lambda: [exe, "webrtc-worker"]
    pool = workers.WorkerPool(2)
    browser = RTCPeerConnection()
    try:
        await pool.ensure()
        if pool.processes() != 2:
            raise SystemExit(f"FAILED  {pool.processes()} of 2 connection processes started")

        channel = browser.createDataChannel("mnp")
        got: list[bytes] = []
        channel.on("message", got.append)
        await browser.setLocalDescription(await browser.createOffer())
        peer = await pool.open(browser.localDescription.sdp)
        ours: list = []
        peer.on("datachannel", ours.append)
        await browser.setRemoteDescription(
            RTCSessionDescription(sdp=await peer.negotiate(), type="answer"))
        await until(lambda: channel.readyState == "open" and ours, "the data channel opens")

        heard: list[bytes] = []
        ours[0].on("message", heard.append)
        channel.send(b"from the browser")
        await until(lambda: heard == [b"from the browser"], "a message reaches the node")

        for _ in range(CHUNKS):
            ours[0].send(CHUNK)
            while ours[0].bufferedAmount > 2_000_000:
                await asyncio.sleep(0.005)
        await until(lambda: sum(map(len, got)) == CHUNKS * len(CHUNK),
                    "what the node sent arrives whole")
        await until(lambda: ours[0].bufferedAmount == 0,
                    "the node's count of what is buffered comes back to zero")
    finally:
        started = [w.process for w in pool._workers]
        await browser.close()
        await pool.close()
    # A process still there would be one an upgrade finds holding
    # meshbay-node.exe. Its exit code says nothing: the node ends a process
    # whose socket has closed, and on Windows that reads 1 however it left.
    if any(p.returncode is None for p in started):
        raise SystemExit("FAILED  a connection process outlived the node's stop")
    return f"{CHUNKS * len(CHUNK) // 1_000_000} MB through a connection process"


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit(__doc__)
    for exe in sys.argv[1:]:
        print(f"OK  {exe}: {asyncio.run(carry(exe))}")


if __name__ == "__main__":
    main()