//! Async client for the Global Socket Relay Network (GSRN).
//!
//! This crate wires the [`gsocket-proto`](::gsocket_proto) codec over a Tokio
//! TCP socket. After the relay matches two peers, the underlying stream
//! becomes a transparent byte pipe — handed off to the TLS-SRP layer
//! (`gsocket-tls-srp`) for end-to-end encryption.
//!
//! Status: relay handshake (LISTEN/CONNECT, START dispatch, STATUS errors,
//! PING/PONG keepalive) is implemented at the protocol level. Full
//! integration with the TLS layer will land alongside the `gsocket-tls-srp`
//! state machine.

pub mod tls;

pub use tls::{
    client_handshake, client_handshake_with_prefix, server_handshake,
    server_handshake_with_prefix, TlsStream,
};

use std::time::Duration;

use gsocket_proto::{
    Accept, Connect, GsAddr, Inbound, Listen, Ping, ProtoFlags, Start, StartRole, Status,
    StatusKind, GSRN_DEFAULT_PORT, GSRN_DEFAULT_PORT_CON, GSRN_PING_INTERVAL_SECS,
    SHORT_PKT_LEN,
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error(transparent)]
    Io(#[from] std::io::Error),
    #[error(transparent)]
    Proto(#[from] gsocket_proto::Error),
    #[error("GSRN returned fatal status: code={0:?} msg={1:?}")]
    Fatal(gsocket_proto::StatusCode, String),
    #[error("GSRN closed the connection before sending START")]
    UnexpectedEof,
    #[error("invalid hostname")]
    BadHost,
}

/// Role to take in the relay handshake.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Role {
    Listen,
    Connect,
}

/// Choose the load-balancing prefix for the GSRN hostname.
///
/// The C ref impl picks `[a-z]` based on the sum of the address bytes.
/// See `GS_ADDR_get_hostname_id` in `gsocket-util.c`.
fn host_prefix_for(addr: &GsAddr) -> char {
    let sum: u32 = addr.as_bytes().iter().map(|&b| b as u32).sum();
    let idx = (sum % 26) as u8;
    (b'a' + idx) as char
}

#[derive(Debug, Clone)]
pub struct GsConfig {
    pub addr: GsAddr,
    pub role: Role,
    pub flags: ProtoFlags,
    /// Hostname template; the load-balancing prefix is prepended.
    /// Defaults to `"gs.thc.org"` matching the public GSRN.
    pub host: String,
    /// LISTEN authentication token. Must be non-zero, and should be REUSED
    /// across re-LISTENs from the same process: GSRN tracks the
    /// (address, token) pair and refuses a re-LISTEN with a *different*
    /// token until `GSRN_TOKEN_LINGER_SEC` seconds have passed (constant
    /// in the C reference). [`GsConfig::new`] generates a fresh random
    /// token; clone the config to reuse it.
    pub token: [u8; gsocket_proto::TOKEN_SIZE],
}

impl GsConfig {
    pub fn new(addr: GsAddr, role: Role) -> Self {
        let mut token = [0u8; gsocket_proto::TOKEN_SIZE];
        rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut token);
        Self {
            addr,
            role,
            flags: ProtoFlags::empty(),
            host: "gs.thc.org".to_string(),
            token,
        }
    }

    fn full_hostname(&self) -> String {
        format!("{}.{}", host_prefix_for(&self.addr), self.host)
    }

    fn port(&self) -> u16 {
        // The C reference defines GSRN_DEFAULT_PORT_CON = 7351 but never reads
        // it — both roles connect to port 443. Match that for interop.
        let _ = GSRN_DEFAULT_PORT_CON;
        GSRN_DEFAULT_PORT
    }
}

/// A successfully handed-off relay stream.
///
/// After `START` is received, the underlying TCP stream is a transparent
/// byte pipe to the peer. Hand this to the TLS-SRP layer.
///
/// `initial_bytes` holds any peer bytes that arrived coalesced with the
/// final GSRN START packet — those must be replayed at the start of the
/// TLS layer's read buffer.
pub struct RelayStream {
    pub role: StartRole,
    pub stream: TcpStream,
    pub initial_bytes: Vec<u8>,
}

/// Drive the GSRN handshake.
///
/// 1. TCP-connect to `<prefix>.<host>:<port>`.
/// 2. Send LISTEN or CONNECT.
/// 3. Loop reading 32-byte messages: PONG (keep-alive ack), STATUS (error),
///    START (peer matched — return the stream).
/// 4. While waiting, send a PING every [`GSRN_PING_INTERVAL_SECS`].
pub async fn handshake(cfg: GsConfig) -> Result<RelayStream, Error> {
    let host = cfg.full_hostname();
    let port = cfg.port();
    tracing::debug!(target: "gsocket", host = %host, port, role = ?cfg.role, "connecting");

    let mut stream = TcpStream::connect((host.as_str(), port)).await?;

    match cfg.role {
        Role::Listen => {
            let mut msg = Listen::new(cfg.addr);
            msg.flags = cfg.flags;
            msg.token = cfg.token;
            stream.write_all(&msg.encode()).await?;
        }
        Role::Connect => {
            let mut msg = Connect::new(cfg.addr);
            msg.flags = cfg.flags;
            stream.write_all(&msg.encode()).await?;
            if cfg.flags.contains(ProtoFlags::FAST_CONNECT) {
                // Skip the START wait per protocol.
                return Ok(RelayStream {
                    role: StartRole::Client,
                    stream,
                    initial_bytes: Vec::new(),
                });
            }
        }
    }

    let ping_interval = Duration::from_secs(GSRN_PING_INTERVAL_SECS);
    let mut ping_timer = tokio::time::interval(ping_interval);
    ping_timer.tick().await; // consume the immediate "burst" tick

    // Persistent accumulator — surviving across select! branches so we
    // don't lose partial reads when the ping timer interrupts us. Using
    // `read_exact` here would be a bug: it is NOT cancellation-safe and
    // would lose partial data on each select cancellation, shifting our
    // packet alignment on the next iteration.
    let mut accum: Vec<u8> = Vec::with_capacity(SHORT_PKT_LEN * 2);
    let mut chunk = [0u8; SHORT_PKT_LEN];

    loop {
        // Drain complete packets from the accumulator first.
        while accum.len() >= SHORT_PKT_LEN {
            let pkt_bytes = {
                let mut tmp = [0u8; SHORT_PKT_LEN];
                tmp.copy_from_slice(&accum[..SHORT_PKT_LEN]);
                tmp
            };
            accum.drain(..SHORT_PKT_LEN);
            match Inbound::decode(&pkt_bytes)? {
                Inbound::Pong(_) => {
                    tracing::trace!(target: "gsocket", "pong");
                }
                Inbound::Status(Status { kind: StatusKind::Fatal, code, msg }) => {
                    let msg_str = trim_nul(&msg);
                    return Err(Error::Fatal(code, msg_str));
                }
                Inbound::Status(s) => {
                    tracing::warn!(target: "gsocket", code = ?s.code, "GSRN warning");
                }
                Inbound::Start(Start { role }) => {
                    // After START, C peer sends ACCEPT back to GSRN. Without
                    // it, GSRN tears down the connection a few seconds later.
                    // See gs_pkt_accept_write() in gsocket-engine.c.
                    stream.write_all(&Accept.encode()).await?;
                    // Anything still in `accum` after START would be peer
                    // bytes that arrived in the same TCP read as the START
                    // packet. Pass them through to the TLS layer.
                    let initial = std::mem::take(&mut accum);
                    return Ok(RelayStream {
                        role,
                        stream,
                        initial_bytes: initial,
                    });
                }
            }
        }

        tokio::select! {
            biased;
            // read() (not read_exact!) is cancellation-safe: if the timer
            // wins this select, no data has been read.
            r = stream.read(&mut chunk) => {
                let n = r?;
                if n == 0 {
                    return Err(Error::UnexpectedEof);
                }
                accum.extend_from_slice(&chunk[..n]);
            }
            _ = ping_timer.tick() => {
                stream.write_all(&Ping.encode()).await?;
            }
        }
    }
}

/// Send an explicit ACCEPT (rarely needed; FAST_CONNECT skips it).
pub async fn send_accept(stream: &mut TcpStream) -> Result<(), Error> {
    stream.write_all(&Accept.encode()).await?;
    Ok(())
}

fn trim_nul(bytes: &[u8]) -> String {
    let end = bytes.iter().position(|&b| b == 0).unwrap_or(bytes.len());
    String::from_utf8_lossy(&bytes[..end]).into_owned()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn host_prefix_is_lowercase_letter() {
        let a = GsAddr([0xff; 16]);
        let c = host_prefix_for(&a);
        assert!(c.is_ascii_lowercase());
    }

    #[test]
    fn full_hostname_format() {
        let cfg = GsConfig::new(GsAddr([0; 16]), Role::Listen);
        let h = cfg.full_hostname();
        assert!(h.starts_with("a."));
        assert!(h.ends_with("gs.thc.org"));
    }

    #[test]
    fn both_roles_use_port_443() {
        // Matches the C reference: GSRN_DEFAULT_PORT_CON is defined but unused.
        let listen = GsConfig::new(GsAddr([0; 16]), Role::Listen);
        let connect = GsConfig::new(GsAddr([0; 16]), Role::Connect);
        assert_eq!(listen.port(), 443);
        assert_eq!(connect.port(), 443);
    }
}
