//! Classic TCP transport (+ optional EPMD lookup).

use crate::error::{Error, Result};
use crate::stream::DistStream;
use erl_dist::epmd::DEFAULT_EPMD_PORT;
use erl_dist::node::NodeName;
use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tracing::debug;

/// TCP carrier helpers.
pub struct TcpTransport;

impl TcpTransport {
    /// Connect to `host:port` and wrap as a [`DistStream`].
    pub async fn connect(host: &str, port: u16) -> Result<DistStream> {
        debug!(%host, port, "tcp connect");
        let stream = TcpStream::connect((host, port)).await?;
        stream.set_nodelay(true)?;
        Ok(DistStream::from_rw(stream))
    }

    /// Connect to a socket address.
    pub async fn connect_addr(addr: SocketAddr) -> Result<DistStream> {
        debug!(%addr, "tcp connect");
        let stream = TcpStream::connect(addr).await?;
        stream.set_nodelay(true)?;
        Ok(DistStream::from_rw(stream))
    }

    /// Resolve a BEAM node name via EPMD, then TCP-connect to its port.
    ///
    /// `node` is a full name like `foo@localhost`.
    pub async fn connect_node(node: &str) -> Result<DistStream> {
        let name: NodeName = node
            .parse()
            .map_err(|e: erl_dist::node::NodeNameError| Error::InvalidName(e.to_string()))?;
        let port = epmd_lookup(name.host(), name.name()).await?;
        Self::connect(name.host(), port).await
    }

    /// Look up a short node name on a host's EPMD.
    pub async fn epmd_port(host: &str, short_name: &str) -> Result<u16> {
        epmd_lookup(host, short_name).await
    }
}

async fn epmd_lookup(host: &str, short_name: &str) -> Result<u16> {
    // Minimal EPMD PORT_PLEASE2 over tokio TCP.
    let mut stream = TcpStream::connect((host, DEFAULT_EPMD_PORT)).await?;
    // PORT_PLEASE2_REQ: 122, then name bytes. Length-prefixed packet (2-byte BE).
    let mut body = Vec::with_capacity(1 + short_name.len());
    body.push(122u8); // PORT_PLEASE2_REQ
    body.extend_from_slice(short_name.as_bytes());
    let mut packet = Vec::with_capacity(2 + body.len());
    packet.extend_from_slice(&(body.len() as u16).to_be_bytes());
    packet.extend_from_slice(&body);
    stream.write_all(&packet).await?;

    // Response is NOT length-prefixed: result(1) [+ port(2) + …] or result≠0 (not found)
    let mut result = [0u8; 1];
    stream.read_exact(&mut result).await?;
    if result[0] != 0 {
        return Err(Error::epmd(format!(
            "node `{short_name}` not registered on EPMD at {host}"
        )));
    }
    let mut port_buf = [0u8; 2];
    stream.read_exact(&mut port_buf).await?;
    let port = u16::from_be_bytes(port_buf);
    debug!(%host, %short_name, port, "epmd lookup ok");
    Ok(port)
}