//! Network Utilities for Citadel Protocol
//!
//! This module provides core networking utilities and types used throughout the
//! Citadel Protocol. It handles network connections, address resolution, and
//! protocol-specific networking operations.
//!
//! # Features
//!
//! - Socket management
//! - Address resolution
//! - Connection handling
//! - Protocol negotiation
//! - Network error handling
//! - Transport selection
//!
//! # Important Notes
//!
//! - Supports TCP and UDP transports
//! - Handles IPv4 and IPv6 addresses
//! - Implements connection timeouts
//! - Provides error recovery
//!
//! # Related Components
//!
//! - `underlying_proto.rs`: Protocol implementation
//! - `udp_internal_interface.rs`: UDP handling
//! - `session.rs`: Session management
//! - `node.rs`: Node implementation

use crate::error::NetworkError;
use crate::proto::node::TlsDomain;
use crate::proto::peer::p2p_conn_handler::generic_error;
use citadel_io::tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use citadel_io::tokio::net::{TcpListener, TcpStream};
use citadel_io::tokio_stream::{Stream, StreamExt};
use citadel_user::serialization::SyncIO;
use citadel_wire::exports::tokio_rustls::{server::TlsStream, TlsAcceptor};
use citadel_wire::exports::{Connection, Endpoint, RecvStream, SendStream};
use citadel_wire::quic::{QuicEndpointListener, QuicNode};
use citadel_wire::tls::TLSQUICInterop;
use futures::{Future, TryStreamExt};
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
use std::fmt::Formatter;
use std::io::Error;
use std::net::SocketAddr;
use std::ops::DerefMut;
use std::path::Path;
use std::pin::Pin;
use std::task::{Context, Poll};

pub enum GenericNetworkStream {
    OrderedReliable(TcpStream),
    OrderedReliableSecure(Box<citadel_wire::exports::tokio_rustls::TlsStream<TcpStream>>),
    // local addr is first addr, remote addr is final addr
    P2P(
        SendStream,
        RecvStream,
        Endpoint,
        Option<Connection>,
        SocketAddr,
    ),
    WebSocket(Box<super::native_websocket::WebSocketByteStream>),
}

impl Unpin for GenericNetworkStream {}

impl Debug for GenericNetworkStream {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let tag = match self {
            Self::OrderedReliable(..) => "OrderedReliable",
            Self::OrderedReliableSecure(..) => "OrderedReliableSecure",
            Self::P2P(..) => "P2P",
            Self::WebSocket(..) => "WebSocket",
        };

        write!(f, "{tag}")
    }
}

impl GenericNetworkStream {
    pub(crate) fn peer_addr(&self) -> std::io::Result<SocketAddr> {
        match self {
            Self::OrderedReliable(stream) => stream.peer_addr(),
            Self::OrderedReliableSecure(stream) => TcpStream::peer_addr(stream.get_ref().0),
            Self::P2P(_, _, _, _, remote_addr) => Ok(*remote_addr),
            Self::WebSocket(ws) => ws.peer_addr(),
        }
    }

    pub(crate) fn local_addr(&self) -> std::io::Result<SocketAddr> {
        match self {
            Self::OrderedReliable(stream) => stream.local_addr(),
            Self::OrderedReliableSecure(stream) => TcpStream::local_addr(stream.get_ref().0),
            Self::P2P(_, _, endpoint, _, _) => endpoint.local_addr(),
            Self::WebSocket(ws) => ws.local_addr(),
        }
    }

    #[allow(dead_code)]
    pub(crate) fn p2p_endpoint(&self) -> Option<Endpoint> {
        match self {
            Self::P2P(_, _, endpoint, _, _) => Some(endpoint.clone()),
            _ => None,
        }
    }

    pub fn take_p2p_connection(&mut self) -> Option<Connection> {
        match self {
            Self::P2P(_, _, _, conn, ..) => conn.take(),
            _ => None,
        }
    }
}

impl AsyncRead for GenericNetworkStream {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        match self.deref_mut() {
            Self::OrderedReliable(stream) => Pin::new(stream).poll_read(cx, buf),
            Self::OrderedReliableSecure(stream) => Pin::new(&mut *stream).poll_read(cx, buf),
            Self::P2P(_, recv, ..) => Pin::new(recv).poll_read(cx, buf),
            Self::WebSocket(ws) => Pin::new(ws).poll_read(cx, buf),
        }
    }
}

impl AsyncWrite for GenericNetworkStream {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, Error>> {
        match self.deref_mut() {
            Self::OrderedReliable(stream) => Pin::new(stream).poll_write(cx, buf),
            Self::OrderedReliableSecure(stream) => Pin::new(&mut *stream).poll_write(cx, buf),
            Self::P2P(sink, ..) => Pin::new(sink).poll_write(cx, buf).map_err(|err| err.into()),
            Self::WebSocket(ws) => Pin::new(ws).poll_write(cx, buf),
        }
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
        match self.deref_mut() {
            Self::OrderedReliable(stream) => Pin::new(stream).poll_flush(cx),
            Self::OrderedReliableSecure(stream) => Pin::new(&mut *stream).poll_flush(cx),
            Self::P2P(sink, ..) => Pin::new(sink).poll_flush(cx),
            Self::WebSocket(ws) => Pin::new(ws).poll_flush(cx),
        }
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
        match self.deref_mut() {
            Self::OrderedReliable(stream) => Pin::new(stream).poll_shutdown(cx),
            Self::OrderedReliableSecure(stream) => Pin::new(&mut *stream).poll_shutdown(cx),
            Self::P2P(sink, ..) => Pin::new(sink).poll_shutdown(cx),
            Self::WebSocket(ws) => Pin::new(ws).poll_shutdown(cx),
        }
    }
}

pub struct GenericNetworkListener {
    future: Pin<Box<dyn StreamOutputImpl>>,
    recv: citadel_io::tokio::sync::mpsc::Receiver<
        std::io::Result<(GenericNetworkStream, SocketAddr)>,
    >,
    local_addr: SocketAddr,
    #[allow(dead_code)]
    quic_endpoint: Option<Endpoint>,
    #[allow(dead_code)]
    redirect_to_quic: Option<(TlsDomain, bool)>,
    tls_domain: TlsDomain,
}

impl GenericNetworkListener {
    pub fn from_quic_node(quic_node: QuicNode, is_self_signed: bool) -> std::io::Result<Self> {
        let endpoint = quic_node.endpoint.clone();
        let local_addr = quic_node.endpoint.local_addr()?;
        let tls_domain = quic_node.tls_domain_opt.clone();
        let mut listener = QuicListener::new(quic_node, is_self_signed);
        let (send, recv) = citadel_io::tokio::sync::mpsc::channel(1024);

        let future = async move {
            while let Some(stream) = listener.next().await {
                let res = stream.map(|(conn, tx, rx, peer_addr, endpoint)| {
                    (
                        GenericNetworkStream::P2P(tx, rx, endpoint, Some(conn), peer_addr),
                        peer_addr,
                    )
                });

                log::trace!(target: "citadel", "RECV raw QUIC stream from {res:?}");
                send.send(res)
                    .await
                    .map_err(|err| generic_error(err.to_string()))?;
            }

            Err(generic_error("QUIC listener died"))
        };

        Ok(Self {
            future: Box::pin(future),
            recv,
            local_addr,
            quic_endpoint: Some(endpoint),
            redirect_to_quic: None,
            tls_domain,
        })
    }

    pub fn new_tcp(
        listener: TcpListener,
        redirect_to_quic: Option<(TlsDomain, bool)>,
    ) -> std::io::Result<Self> {
        let (inbound_streams_tx, recv) = citadel_io::tokio::sync::mpsc::channel(1024);
        let local_addr = listener.local_addr()?;
        let tls_domain = redirect_to_quic.as_ref().and_then(|r| r.0.clone());

        let future = async move {
            let redirect_to_quic = &redirect_to_quic;
            loop {
                let (stream, addr) = listener.accept().await?;
                log::trace!(target: "citadel", "Received raw TCP stream from {addr:?}: {stream:?}");

                // ensures that any errors do not terminate the listener as a whole
                async fn handle_stream_non_terminating(
                    stream: TcpStream,
                    addr: SocketAddr,
                    redirect_to_quic: &Option<(TlsDomain, bool)>,
                ) -> std::io::Result<(GenericNetworkStream, SocketAddr)> {
                    // Disable Nagle on every accepted TCP stream (not just the QUIC-redirect
                    // handshake): the long-lived OrderedReliable data path needs low per-message
                    // latency, and write-coalescing only adds delay.
                    stream.set_nodelay(true)?;
                    let first_packet = if let Some((domain, is_self_signed)) = redirect_to_quic {
                        FirstPacket::P2P {
                            domain: domain.clone(),
                            external_addr: addr,
                            is_self_signed: *is_self_signed,
                        }
                    } else {
                        FirstPacket::OrderedReliable {
                            external_addr: addr,
                        }
                    };

                    let conn = super::write_one_packet(
                        stream,
                        first_packet
                            .serialize_to_vector()
                            .map_err(|err| generic_error(err.into_string()))?,
                    )
                    .await
                    .map_err(|err| generic_error(err.to_string()))?;
                    Ok((GenericNetworkStream::OrderedReliable(conn), addr))
                }

                let redirect_to_quic = redirect_to_quic.clone();
                let inbound_streams_tx = inbound_streams_tx.clone();
                let handle_stream = async move {
                    if let Err(err) = inbound_streams_tx
                        .send(handle_stream_non_terminating(stream, addr, &redirect_to_quic).await)
                        .await
                        .map_err(|err| generic_error(err.to_string()))
                    {
                        log::error!(target: "citadel", "Error sending inbound stream from {addr} to listener: {err}");
                    }
                };

                // Spawn to prevent backpressure against pending inbound connections
                spawn!(handle_stream);
            }
        };

        Ok(Self {
            future: Box::pin(future),
            recv,
            local_addr,
            quic_endpoint: None,
            redirect_to_quic: None,
            tls_domain,
        })
    }

    pub fn new_tls(mut listener: TlsListener) -> std::io::Result<Self> {
        let (send, recv) = citadel_io::tokio::sync::mpsc::channel(1024);
        let local_addr = listener.local_addr;
        let tls_domain = listener.tls_domain.clone();

        let future = async move {
            loop {
                let next_connection = listener
                    .next()
                    .await
                    .ok_or_else(|| generic_error("TLS listener died"))?;

                match next_connection {
                    Ok((stream, addr)) => {
                        log::trace!(target: "citadel", "Received raw TLS stream from {addr:?}: {stream:?}");
                        if let Err(err) = send
                            .send(Ok((
                                GenericNetworkStream::OrderedReliableSecure(Box::new(
                                    stream.into(),
                                )),
                                addr,
                            )))
                            .await
                        {
                            log::error!(target: "citadel", "Failed to send TLS handshake packet: {err:?}");
                        }
                    }
                    Err(err) => {
                        log::debug!(target: "citadel", "TLS Stream died: {err:?}")
                    }
                }
            }
        };

        Ok(Self {
            future: Box::pin(future),
            recv,
            local_addr,
            quic_endpoint: None,
            redirect_to_quic: None,
            tls_domain,
        })
    }

    pub fn local_addr(&self) -> std::io::Result<SocketAddr> {
        Ok(self.local_addr)
    }

    #[allow(dead_code)]
    pub fn tls_domain(&self) -> TlsDomain {
        self.tls_domain.clone()
    }

    /// For P2P connections, this should exist
    #[allow(dead_code)]
    pub fn quic_endpoint(&self) -> Option<Endpoint> {
        self.quic_endpoint.clone()
    }

    /// Create a WebSocket listener that accepts HTTP upgrade requests.
    ///
    /// Incoming TCP connections are upgraded to WebSocket, then wrapped as
    /// `GenericNetworkStream::WebSocket`. No `FirstPacket` is sent — the
    /// HTTP upgrade itself serves as protocol negotiation. WASM clients
    /// (the primary WebSocket consumers) don't read `FirstPacket`.
    pub fn new_websocket(listener: TcpListener) -> std::io::Result<Self> {
        let local_addr = listener.local_addr()?;
        let (send, recv) = citadel_io::tokio::sync::mpsc::channel(1024);

        let future = async move {
            loop {
                let (tcp_stream, addr) = listener.accept().await?;
                // Disable Nagle before the WebSocket upgrade: framed packets need low latency.
                tcp_stream.set_nodelay(true)?;
                let send = send.clone();

                let handle_ws = async move {
                    match tokio_tungstenite::accept_async(tcp_stream).await {
                        Ok(ws_stream) => {
                            let byte_stream = super::native_websocket::WebSocketByteStream::new(
                                ws_stream, addr, addr,
                            );
                            let generic = GenericNetworkStream::WebSocket(Box::new(byte_stream));
                            if let Err(e) = send.send(Ok((generic, addr))).await {
                                log::error!(target: "citadel", "Failed to send WebSocket stream: {e}");
                            }
                        }
                        Err(e) => {
                            log::debug!(target: "citadel", "WebSocket handshake failed from {addr}: {e}");
                        }
                    }
                    Ok::<(), std::io::Error>(())
                };

                // Spawn to prevent backpressure against pending inbound connections
                spawn!(handle_ws);
            }
        };

        Ok(Self {
            future: Box::pin(future),
            recv,
            local_addr,
            quic_endpoint: None,
            redirect_to_quic: None,
            tls_domain: None,
        })
    }
}

impl Stream for GenericNetworkListener {
    type Item = std::io::Result<(GenericNetworkStream, SocketAddr)>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let Self { future, recv, .. } = &mut *self;

        // if this future ends, it's over
        match future.as_mut().poll(cx) {
            Poll::Pending => {}
            Poll::Ready(res) => {
                // assert err
                log::warn!(target: "citadel", "ERR: {res:?}");
                return Poll::Ready(Some(Err(res.unwrap_err())));
            }
        }

        Pin::new(recv).poll_recv(cx)
    }
}

pub struct TlsListener {
    future: Pin<Box<dyn StreamOutputImpl>>,
    recv: citadel_io::tokio::sync::mpsc::Receiver<
        std::io::Result<(TlsStream<TcpStream>, SocketAddr)>,
    >,
    local_addr: SocketAddr,
    tls_domain: TlsDomain,
}

impl TlsListener {
    pub fn new(
        inner: TcpListener,
        tls_acceptor: TlsAcceptor,
        domain: TlsDomain,
        is_self_signed: bool,
    ) -> std::io::Result<Self> {
        // TODO: add channel capacity for acceptors
        let (send, recv) = citadel_io::tokio::sync::mpsc::channel(1024);
        let local_addr = inner.local_addr()?;
        let tls_domain = domain.clone();

        let future = async move {
            let tls_acceptor = &tls_acceptor;
            let domain = &domain;
            let send = &send;

            let acceptor_stream = async_stream::stream! {
                    loop {
                        yield inner.accept().await
                    }
            };

            acceptor_stream.try_for_each_concurrent(None, |(stream, addr)| async move {
                log::trace!(target: "citadel", "TLs-listener RECV Raw TCP stream from {addr:?} : {stream:?}");
                let domain = domain.clone();

                async fn handle_stream_non_terminating(stream: TcpStream, addr: SocketAddr, domain: TlsDomain, is_self_signed: bool, tls_acceptor: &TlsAcceptor) -> std::io::Result<(TlsStream<TcpStream>, SocketAddr)> {
                    // Disable Nagle before the TLS upgrade: the encrypted data path carries small
                    // framed packets and needs low per-message latency.
                    stream.set_nodelay(true)?;
                    let serialized_first_packet = FirstPacket::OrderedReliableSecure { domain, external_addr: addr, is_self_signed }.serialize_to_vector().map_err(|err| generic_error(err.into_string()))?;
                    let stream = super::write_one_packet(stream, serialized_first_packet).await.map_err(|err| generic_error(err.into_string()))?;
                    // Upgrade TCP stream to TLS stream
                    tls_acceptor.accept(stream).await.map(|r| (r, addr))
                }

                send.send(handle_stream_non_terminating(stream, addr, domain, is_self_signed, tls_acceptor).await).await.map_err(|err| generic_error(err.to_string()))
            }).await
        };

        Ok(Self {
            future: Box::pin(future),
            recv,
            local_addr,
            tls_domain,
        })
    }

    /// Given a path and password, returns the asymmetric crypto identity
    /// Also stores the QUIC keys for use in case of P2P
    #[allow(dead_code)]
    pub fn load_tls_pkcs<P: AsRef<Path>, T: AsRef<str>>(
        path: P,
        password: T,
    ) -> Result<TLSQUICInterop, NetworkError> {
        let bytes = std::fs::read(path).map_err(|err| NetworkError::generic(err.to_string()))?;
        citadel_wire::tls::create_server_config(&bytes, password.as_ref())
            .map_err(|err| NetworkError::generic(err.to_string()))
    }
}

impl Stream for TlsListener {
    type Item = std::io::Result<(TlsStream<TcpStream>, SocketAddr)>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let Self { future, recv, .. } = &mut *self;

        // if this future ends, it's over
        match future.as_mut().poll(cx) {
            Poll::Pending => {}
            Poll::Ready(res) => {
                // assert err
                log::warn!(target: "citadel", "ERR: {res:?}");
                return Poll::Ready(Some(Err(res.unwrap_err())));
            }
        }

        Pin::new(recv).poll_recv(cx)
    }
}

pub struct QuicListener {
    future: Pin<Box<dyn StreamOutputImpl>>,
    recv: citadel_io::tokio::sync::mpsc::Receiver<std::io::Result<IncomingQuicConnection>>,
    #[allow(dead_code)]
    is_self_signed: bool,
}

type IncomingQuicConnection = (Connection, SendStream, RecvStream, SocketAddr, Endpoint);

impl QuicListener {
    pub fn new(mut server: QuicNode, is_self_signed: bool) -> Self {
        let (send, recv) = citadel_io::tokio::sync::mpsc::channel(1024);
        let endpoint = server.endpoint.clone();

        let future = async move {
            loop {
                let server = &mut server;

                let acceptor_stream = async_stream::stream! {
                    loop {
                        yield server.next_connection().await.map_err(|err| generic_error(err.to_string()));
                    }
                };

                let endpoint = &endpoint;
                let send = &send;

                acceptor_stream
                    .try_for_each_concurrent(None, |(conn, tx, rx)| async move {
                        let addr = conn.remote_address();
                        log::trace!(target: "citadel", "RECV {:?} from {:?}", &conn, addr);
                        send.send(Ok((conn, tx, rx, addr, endpoint.clone())))
                            .await
                            .map_err(generic_error)
                    })
                    .await?;
            }
        };

        Self {
            future: Box::pin(future),
            recv,
            is_self_signed,
        }
    }
}

impl Stream for QuicListener {
    type Item = std::io::Result<(Connection, SendStream, RecvStream, SocketAddr, Endpoint)>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let Self { future, recv, .. } = &mut *self;

        // if this future ends, it's over
        match future.as_mut().poll(cx) {
            Poll::Pending => {}
            Poll::Ready(res) => {
                // assert err
                log::warn!(target: "citadel", "ERR: {res:?}");
                return Poll::Ready(Some(Err(res.unwrap_err())));
            }
        }

        Pin::new(recv).poll_recv(cx)
    }
}

#[derive(Serialize, Deserialize)]
pub enum FirstPacket {
    OrderedReliable {
        external_addr: SocketAddr,
    },
    OrderedReliableSecure {
        domain: TlsDomain,
        external_addr: SocketAddr,
        is_self_signed: bool,
    },
    P2P {
        domain: TlsDomain,
        external_addr: SocketAddr,
        is_self_signed: bool,
    },
}

pub struct DualListener {
    future: Pin<Box<dyn StreamOutputImpl>>,
    recv: citadel_io::tokio::sync::mpsc::Receiver<
        std::io::Result<(GenericNetworkStream, SocketAddr)>,
    >,
}

impl DualListener {
    /// Merge a signaling listener (output suppressed) with a target listener
    /// (output forwarded). Used by [`ProtocolUpgrade`] implementations.
    pub fn merge_suppressing_signal(
        mut signal_listener: DualListener,
        mut target_listener: DualListener,
    ) -> Self {
        let (tx, recv) = citadel_io::tokio::sync::mpsc::channel(1024);

        let future = async move {
            let signal_task = async move {
                loop {
                    match signal_listener.next().await {
                        Some(Ok(_)) => { /* signaled connection, silently drop */ }
                        Some(Err(e)) => return Err(e),
                        None => return Err(generic_error("Signal listener ended")),
                    }
                }
            };

            let target_task = async move {
                loop {
                    match target_listener.next().await {
                        Some(res) => tx
                            .send(res)
                            .await
                            .map_err(|err| generic_error(err.to_string()))?,
                        None => return Err::<(), _>(generic_error("Target listener ended")),
                    }
                }
            };

            citadel_io::tokio::try_join!(signal_task, target_task).map(|_: ((), ())| ())
        };

        Self {
            future: Box::pin(future),
            recv,
        }
    }

    pub fn new(
        mut tcp_or_tls_listener: GenericNetworkListener,
        quic_listener: Option<GenericNetworkListener>,
    ) -> Self {
        let (tx, recv) = citadel_io::tokio::sync::mpsc::channel(1024);
        let tx2 = tx.clone();
        let redirects_to_quic = quic_listener.is_some();

        let future = async move {
            let tcp_or_tls_listener_future = async move {
                loop {
                    match tcp_or_tls_listener.next().await {
                        Some(res) => {
                            let res = res;
                            // only return value IF the dual listener only returns quic streams
                            if !redirects_to_quic {
                                tx.send(res)
                                    .await
                                    .map_err(|err| generic_error(err.to_string()))?
                            }
                        }
                        None => return Err::<(), _>(generic_error("Tcp_or_tls stream died")),
                    }
                }
            };

            let quic_listener_future = async move {
                if let Some(mut quic_listener) = quic_listener {
                    loop {
                        match quic_listener.next().await {
                            Some(res) => tx2
                                .send(res)
                                .await
                                .map_err(|err| generic_error(err.to_string()))?,
                            None => return Err::<(), _>(generic_error("Tcp_or_tls stream died")),
                        }
                    }
                } else {
                    Ok(())
                }
            };

            citadel_io::tokio::try_join!(tcp_or_tls_listener_future, quic_listener_future)
                .map(|_| ())
        };

        Self {
            future: Box::pin(future),
            recv,
        }
    }

    /// Merge an additional `GenericNetworkListener` into this `DualListener`.
    ///
    /// Returns a new `DualListener` that yields connections from BOTH the
    /// existing listener and the additional one. Used to add optional
    /// WebSocket listeners alongside the primary transport listener.
    pub fn merge_with(mut self, mut additional: GenericNetworkListener) -> Self {
        let (tx, recv) = citadel_io::tokio::sync::mpsc::channel(1024);
        let tx2 = tx.clone();

        let future = async move {
            let existing_task = async move {
                loop {
                    match self.next().await {
                        Some(res) => tx
                            .send(res)
                            .await
                            .map_err(|e| generic_error(e.to_string()))?,
                        None => return Err::<(), _>(generic_error("Primary listener ended")),
                    }
                }
            };

            let additional_task = async move {
                loop {
                    match additional.next().await {
                        Some(res) => tx2
                            .send(res)
                            .await
                            .map_err(|e| generic_error(e.to_string()))?,
                        None => return Err::<(), _>(generic_error("Additional listener ended")),
                    }
                }
            };

            citadel_io::tokio::try_join!(existing_task, additional_task).map(|_: ((), ())| ())
        };

        Self {
            future: Box::pin(future),
            recv,
        }
    }
}

impl Stream for DualListener {
    type Item = std::io::Result<(GenericNetworkStream, SocketAddr)>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let Self { future, recv, .. } = &mut *self;

        // If this future ends, it's over
        match future.as_mut().poll(cx) {
            Poll::Pending => {}
            Poll::Ready(res) => {
                // assert err
                log::warn!(target: "citadel", "ERR: {res:?}");
                return Poll::Ready(Some(Err(res.unwrap_err())));
            }
        }

        Pin::new(recv).poll_recv(cx)
    }
}

// Always require Send so that listener types satisfy ProtocolIO::Listener bounds.
// All futures stored here capture only Send types (TcpListener, channels, etc.).
trait StreamOutputImpl: Future<Output = std::io::Result<()>> + Send + 'static {}
impl<T: Future<Output = std::io::Result<()>> + Send + 'static> StreamOutputImpl for T {}
