//! WebSocket bridge: exposes the Minecraft TCP port as a WebSocket endpoint.
//!
//! The Cloudflare tunnel points at this bridge instead of the raw Minecraft
//! port, because Cloudflare's edge speaks HTTP(S) only. A player runs
//! `mchost connect <tunnel-url>` on their machine; that helper opens a
//! WebSocket to the tunnel and listens on a local TCP port, so a vanilla
//! Minecraft client can join at `127.0.0.1:25565` with no mods.

use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use futures_util::{SinkExt, StreamExt};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
use tokio_tungstenite::tungstenite::http::Response as HttpResponseBuilt;

use crate::encoding::percent_decode;
use crate::ui;

pub const BRIDGE_PORT_DEFAULT: u16 = 8081;

/// Bytes are forwarded to the tunnel in chunks of this size.
const CHUNK: usize = 32 * 1024;

pub struct Bridge {
    stop: Arc<AtomicBool>,
}

impl Bridge {
    pub fn shutdown(&self) {
        self.stop.store(true, Ordering::SeqCst);
    }
}

pub fn spawn(mc_port: u16, bridge_port: u16, secret: Option<String>) -> Bridge {
    let stop = Arc::new(AtomicBool::new(false));
    let thread_stop = Arc::clone(&stop);
    std::thread::spawn(move || {
        let runtime = match tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
        {
            Ok(runtime) => runtime,
            Err(e) => {
                ui::error(&format!("bridge: could not start its runtime: {e}"));
                return;
            }
        };
        runtime.block_on(serve(mc_port, bridge_port, secret, thread_stop));
    });
    Bridge { stop }
}

async fn serve(mc_port: u16, bridge_port: u16, secret: Option<String>, stop: Arc<AtomicBool>) {
    let listener = match TcpListener::bind(("127.0.0.1", bridge_port)).await {
        Ok(listener) => listener,
        Err(e) => {
            ui::error(&format!(
                "bridge: cannot listen on 127.0.0.1:{bridge_port}: {e}"
            ));
            return;
        }
    };
    ui::bridge_ok(&format!(
        "listening on 127.0.0.1:{bridge_port} -> minecraft 127.0.0.1:{mc_port}{}",
        if secret.is_some() {
            " (secret required)"
        } else {
            ""
        }
    ));

    loop {
        // Bounded wait so the stop flag is noticed promptly.
        let accepted = tokio::time::timeout(Duration::from_millis(250), listener.accept()).await;
        let (stream, peer) = match accepted {
            Ok(Ok(pair)) => pair,
            Ok(Err(e)) => {
                ui::warn(&format!("bridge: accept failed: {e}"));
                continue;
            }
            Err(_) => {
                if stop.load(Ordering::SeqCst) {
                    break;
                }
                continue;
            }
        };
        let secret = secret.clone();
        tokio::spawn(async move {
            handle(stream, peer, mc_port, secret).await;
        });
    }
    ui::bridge("stopped");
}

/// The callback's error type (an HTTP response) is large; nothing we can do
/// about that, it is tungstenite's signature.
#[allow(clippy::result_large_err)]
async fn handle(stream: TcpStream, peer: SocketAddr, mc_port: u16, secret: Option<String>) {
    // WebSocket handshake; reject clients that do not present the shared secret.
    let expected = secret.clone();
    let ws = match tokio_tungstenite::accept_hdr_async(
        stream,
        move |request: &Request, response: Response| {
            if secret_matches(request.uri().query(), expected.as_deref()) {
                Ok(response)
            } else {
                Err(unauthorized())
            }
        },
    )
    .await
    {
        Ok(ws) => ws,
        Err(e) => {
            ui::warn(&format!("bridge: handshake from {peer} rejected: {e}"));
            return;
        }
    };

    let mc = match tokio::time::timeout(
        Duration::from_secs(5),
        TcpStream::connect(("127.0.0.1", mc_port)),
    )
    .await
    {
        Ok(Ok(mc)) => mc,
        Ok(Err(e)) => {
            ui::warn(&format!(
                "bridge: could not reach the minecraft server on 127.0.0.1:{mc_port}: {e}"
            ));
            return;
        }
        Err(_) => {
            ui::warn("bridge: timed out connecting to the minecraft server");
            return;
        }
    };

    ui::bridge_ok(&format!("player connected through the tunnel ({peer})"));
    match pump(ws, mc).await {
        Ok(()) => ui::bridge(&format!("player disconnected ({peer})")),
        // A vanished far end (no WebSocket close frame) is just a disconnect.
        Err(e) if e.contains("reset without closing handshake") => {
            ui::bridge(&format!("player disconnected ({peer})"))
        }
        Err(e) => ui::bridge(&format!("player connection ({peer}) ended: {e}")),
    }
}

/// Pump bytes between a WebSocket session and the local Minecraft TCP socket.
///
/// Used by both sides of the tunnel: the server-side bridge (plain TCP) and
/// `mchost connect` (TLS).
pub(crate) async fn pump<S>(ws: WebSocketStream<S>, mc: TcpStream) -> Result<(), String>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    let (mut ws_sink, mut ws_stream) = ws.split();
    let (mut mc_reader, mut mc_writer) = mc.into_split();
    let mut buf = vec![0u8; CHUNK];

    loop {
        tokio::select! {
            message = ws_stream.next() => match message {
                Some(Ok(Message::Binary(data))) => {
                    mc_writer.write_all(&data).await.map_err(|e| e.to_string())?;
                }
                Some(Ok(Message::Text(text))) => {
                    mc_writer.write_all(text.as_bytes()).await.map_err(|e| e.to_string())?;
                }
                Some(Ok(Message::Ping(_))) => {
                    // tungstenite queues the pong reply; flushing sends it.
                    ws_sink.flush().await.map_err(|e| e.to_string())?;
                }
                Some(Ok(Message::Pong(_))) | Some(Ok(Message::Frame(_))) => {}
                Some(Ok(Message::Close(_))) | None => return Ok(()),
                Some(Err(e)) => return Err(e.to_string()),
            },
            read = mc_reader.read(&mut buf) => match read {
                Ok(0) => return Ok(()),
                Ok(n) => {
                    ws_sink
                        .send(Message::Binary(buf[..n].to_vec().into()))
                        .await
                        .map_err(|e| e.to_string())?;
                }
                Err(e) => return Err(e.to_string()),
            },
        }
    }
}

/// The bridge accepts everyone unless a secret was configured.
fn secret_matches(query: Option<&str>, expected: Option<&str>) -> bool {
    let Some(expected) = expected else {
        return true;
    };
    query
        .and_then(|query| {
            query.split('&').find_map(|pair| {
                pair.strip_prefix("secret=")
                    .or_else(|| pair.strip_prefix("s="))
            })
        })
        .is_some_and(|found| percent_decode(found) == expected)
}

fn unauthorized() -> ErrorResponse {
    HttpResponseBuilt::builder()
        .status(401)
        .body(Some(
            "this bridge requires a secret; run: mchost connect <url> --secret <secret>"
                .to_string(),
        ))
        .unwrap_or_default()
}