//! One gateway session: a browser socket, one upstream TLS connection and one
//! `ClientCore`.
//!
//! The browser sends `Command` JSON and renders `View` JSON. All game state,
//! protocol framing and balance arithmetic stay in the Rust core.

use std::collections::HashMap;
use std::net::TcpStream;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use std::time::{Duration, Instant};

use rusqlite::Connection;
use serde::{Deserialize, Serialize};
use tungstenite::error::Error as WsError;
use tungstenite::http::HeaderValue;
use tungstenite::http::header::ORIGIN;
use tungstenite::protocol::{WebSocket, WebSocketConfig};
use tungstenite::{Message, accept_hdr_with_config};

use zagan_bet::client::core::{ClientCore, Effect};
use zagan_bet::client::database;
use zagan_bet::client::transport::{self, TlsStream};
use zagan_bet::client::view::{Command, View};
use zagan_bet::protocol::Packet;
use zagan_bet::version::VERSION;

use crate::config::Config;

const COOKIE_NAME: &str = "zagan_session";
/// On HTTPS, use the `__Host-` prefix so a sibling origin cannot set a
/// domain-scoped cookie and fixate the session.
const HOST_COOKIE_NAME: &str = "__Host-zagan_session";
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
const POLL: Duration = Duration::from_millis(10);

pub enum SocketEvent {
    Attach(WebSocket<TcpStream>),
}

/// Session id → generation and the thread that owns it.
///
/// The generation lets an ending thread remove its own entry without deleting
/// a replacement session that reused the same cookie.
pub type Registry = Arc<Mutex<HashMap<String, (u64, mpsc::Sender<SocketEvent>)>>>;

static NEXT_GENERATION: AtomicU64 = AtomicU64::new(1);

pub fn new_registry() -> Registry {
    Arc::new(Mutex::new(HashMap::new()))
}

/// Complete the WebSocket handshake, then attach this socket to an existing
/// session or start one.
pub fn accept(stream: TcpStream, config: Config, registry: Registry) -> Result<(), String> {
    let mut session_id = None;
    let origin = config.origin.clone();
    let secure = config.cookie_secure;

    let ws_config = WebSocketConfig {
        // A command or view is a few KiB at most; keep a compromised page from
        // making the gateway buffer tens of megabytes.
        max_message_size: Some(64 * 1024),
        max_frame_size: Some(64 * 1024),
        ..WebSocketConfig::default()
    };
    let mut ws = accept_hdr_with_config(
        stream,
        |request: &tungstenite::handshake::server::Request,
         mut response: tungstenite::handshake::server::Response| {
            if let Some(header) = request.headers().get(ORIGIN) {
                if header.to_str().ok() != Some(origin.as_str()) {
                    return Err(tungstenite::http::Response::builder()
                        .status(403)
                        .body(Some("origin not allowed".to_string()))
                        .expect("static response"));
                }
            }

            let found = request
                .headers()
                .get("cookie")
                .and_then(|value| value.to_str().ok())
                .and_then(cookie_id);
            let (id, fresh) = match found {
                Some(id) => (id, false),
                None => (random_id(), true),
            };
            if fresh {
                let mut cookie = format!("{COOKIE_NAME}={id}; Path=/; HttpOnly; SameSite=Strict");
                if secure {
                    cookie.push_str("; Secure");
                }
                if let Ok(value) = HeaderValue::from_str(&cookie) {
                    response.headers_mut().append("set-cookie", value);
                }
            }
            session_id = Some(id);
            Ok(response)
        },
        Some(ws_config),
    )
    .map_err(|error| format!("WebSocket handshake failed: {error}"))?;

    let id = session_id.ok_or_else(|| "handshake produced no session id".to_string())?;
    ws.get_ref()
        .set_nonblocking(true)
        .map_err(|error| format!("could not make the browser socket nonblocking: {error}"))?;

    let (sender, receiver) = mpsc::channel();
    let generation = NEXT_GENERATION.fetch_add(1, Ordering::Relaxed);
    {
        let mut sessions = registry.lock().map_err(|_| "session registry poisoned")?;
        if let Some((_, existing)) = sessions.get(&id).cloned() {
            match existing.send(SocketEvent::Attach(ws)) {
                Ok(()) => return Ok(()),
                Err(mpsc::SendError(SocketEvent::Attach(returned))) => {
                    sessions.remove(&id);
                    ws = returned;
                }
            }
        }
        sessions.insert(id.clone(), (generation, sender.clone()));
    }

    thread::spawn(move || {
        let session = match Session::new(id.clone(), config) {
            Ok(session) => Some(session),
            Err(error) => {
                eprintln!("[webd {id}] could not start: {error}");
                None
            }
        };
        if let Some(mut session) = session {
            session.attach(ws);
            session.poll_loop(&receiver);
        }
        if let Ok(mut sessions) = registry.lock() {
            let own = sessions
                .get(&id)
                .map(|(generation_now, _)| *generation_now == generation)
                .unwrap_or(false);
            if own {
                sessions.remove(&id);
            }
        }
        eprintln!("[webd {id}] session ended");
    });

    Ok(())
}

fn cookie_id(raw: &str) -> Option<String> {
    for pair in raw.split(';') {
        let (name, value) = pair.split_once('=')?;
        if matches!(name.trim(), COOKIE_NAME | HOST_COOKIE_NAME) {
            let value = value.trim();
            if value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
                return Some(value.to_ascii_lowercase());
            }
            return None;
        }
    }
    None
}

fn random_id() -> String {
    use rand::RngCore;
    let mut bytes = [0u8; 32];
    rand::thread_rng().fill_bytes(&mut bytes);
    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}

#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ClientMessage {
    Command { command: Command },
    Ping,
}

#[derive(Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ServerMessage<'a> {
    Hello {
        version: &'a str,
        grace_seconds: u64,
        /// Signed-in user, so a refresh during a game can skip the login card.
        user: Option<&'a str>,
        /// Last known balance, for views that do not carry one.
        balance: Option<i64>,
        /// Estimated game-server Unix time, milliseconds.
        time: u64,
    },
    View {
        view: &'a View,
    },
    Notice {
        text: &'a str,
    },
    Error {
        text: &'a str,
    },
    Replaced,
    Pong,
}

struct Session {
    id: String,
    config: Config,
    core: ClientCore,
    /// In-memory trust store seeded with the pinned fingerprint. No player
    /// credentials are ever stored here.
    trust: Connection,
    upstream: Option<TlsStream>,
    ws: Option<WebSocket<TcpStream>>,
    pending_id: Option<u64>,
    pending_since: Option<Instant>,
    grace_until: Option<Instant>,
    drop_upstream: bool,
}

impl Session {
    fn new(id: String, config: Config) -> Result<Self, String> {
        let trust = Connection::open_in_memory()
            .map_err(|error| format!("could not open the trust store: {error}"))?;
        database::init_db(&trust);
        database::save_fingerprint(&trust, &config.upstream, &config.fingerprint)
            .map_err(|error| format!("could not pin the upstream key: {error}"))?;

        Ok(Self {
            id,
            config,
            core: ClientCore::new(),
            trust,
            upstream: None,
            ws: None,
            pending_id: None,
            pending_since: None,
            grace_until: None,
            drop_upstream: false,
        })
    }

    fn attach(&mut self, ws: WebSocket<TcpStream>) {
        if let Some(mut old) = self.ws.take() {
            send_json(&mut old, &ServerMessage::Replaced);
            close_socket(&mut old);
        }
        eprintln!("[webd {}] browser attached", self.id);
        self.ws = Some(ws);
        self.grace_until = None;
        let mut ws = self.ws.take().expect("just attached");
        if let Err(text) = self.ensure_upstream() {
            send_json(&mut ws, &ServerMessage::Error { text: &text });
        }
        send_json(
            &mut ws,
            &ServerMessage::Hello {
                version: VERSION,
                grace_seconds: self.config.grace.as_secs(),
                user: self.core.user(),
                balance: self.core.balance(),
                time: self.core.server_now_ms(),
            },
        );
        self.send_view(&mut ws);
        self.ws = Some(ws);
    }

    fn poll_loop(&mut self, receiver: &mpsc::Receiver<SocketEvent>) {
        loop {
            match receiver.recv_timeout(POLL) {
                Ok(SocketEvent::Attach(ws)) => self.attach(ws),
                Err(mpsc::RecvTimeoutError::Timeout) => {}
                Err(mpsc::RecvTimeoutError::Disconnected) => break,
            }

            self.pump_browser();
            self.pump_upstream();
            self.check_request_timeout();

            if self.ws.is_some() {
                self.grace_until = None;
            } else if let Some(deadline) = self.grace_until {
                if Instant::now() >= deadline {
                    break;
                }
            } else {
                self.grace_until = Some(Instant::now() + self.config.grace);
            }
        }

        // Dropping the TLS stream closes upstream; the server forfeits any
        // live hand, exactly like a desktop client disconnecting.
        self.upstream = None;
        eprintln!("[webd {}] grace expired", self.id);
    }

    fn pump_browser(&mut self) {
        let Some(mut ws) = self.ws.take() else {
            return;
        };
        let mut keep = true;

        loop {
            match ws.read() {
                Ok(Message::Text(text)) => {
                    if !self.handle_browser_text(&text, &mut ws) {
                        keep = false;
                        break;
                    }
                }
                Ok(Message::Close(_)) => {
                    keep = false;
                    break;
                }
                Ok(Message::Ping(payload)) => {
                    let _ = send_message(&mut ws, Message::Pong(payload));
                }
                Ok(Message::Binary(_)) | Ok(Message::Pong(_)) | Ok(Message::Frame(_)) => {}
                Err(WsError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock => break,
                Err(WsError::ConnectionClosed) | Err(WsError::AlreadyClosed) => {
                    keep = false;
                    break;
                }
                Err(_) => {
                    keep = false;
                    break;
                }
            }
        }

        if keep {
            self.ws = Some(ws);
        } else {
            self.grace_until = Some(Instant::now() + self.config.grace);
            self.pending_id = None;
            self.pending_since = None;
        }
    }

    fn handle_browser_text(&mut self, text: &str, ws: &mut WebSocket<TcpStream>) -> bool {
        let message = match serde_json::from_str::<ClientMessage>(text) {
            Ok(message) => message,
            Err(error) => {
                send_json(
                    ws,
                    &ServerMessage::Error {
                        text: &format!("Malformed message: {error}"),
                    },
                );
                return true;
            }
        };

        match message {
            ClientMessage::Ping => send_json(ws, &ServerMessage::Pong),
            ClientMessage::Command { command } => {
                if let Err(text) = self.ensure_upstream() {
                    send_json(ws, &ServerMessage::Error { text: &text });
                    self.send_view(ws);
                    return true;
                }
                let effects = self.core.command(command);
                self.apply_effects(effects, Some(ws));
            }
        }
        true
    }

    fn ensure_upstream(&mut self) -> Result<(), String> {
        if self.upstream.is_some() {
            return Ok(());
        }

        let mut trust = |_prompt: &transport::TrustPrompt| false;
        match transport::connect(&self.config.upstream, &self.trust, false, &mut trust) {
            Ok(connected) => {
                self.core.set_connected(true);
                self.core.set_server_time(connected.server_time);
                self.core.set_host_info(
                    Some(self.config.upstream.clone()),
                    Some(self.config.fingerprint.clone()),
                );
                self.upstream = Some(connected.stream);
                eprintln!(
                    "[webd {}] upstream connected: {} ({}), ping {:.1}ms",
                    self.id,
                    self.config.upstream,
                    connected.server_version,
                    connected.elapsed.as_secs_f64() * 1000.0
                );
                Ok(())
            }
            Err(error) => {
                self.core.reset_after_disconnect();
                Err(format!("Upstream connection failed: {error}"))
            }
        }
    }

    fn pump_upstream(&mut self) {
        let Some(mut upstream) = self.upstream.take() else {
            return;
        };
        let mut ws = self.ws.take();
        let mut disconnected = None;

        loop {
            match upstream.read_available_packet() {
                Ok(Some(Packet::Event { body, .. })) => {
                    if let Some(effects) = self.core.handle_async_line(&body) {
                        self.apply_effects(effects, ws.as_mut());
                    }
                }
                Ok(Some(Packet::Reply { id, body })) => {
                    if self.pending_id == Some(id) {
                        self.pending_id = None;
                        self.pending_since = None;
                    }
                    let effects = self.core.handle_reply_id(id, &body);
                    self.apply_effects(effects, ws.as_mut());
                }
                Ok(None) => break,
                Err(error) => {
                    disconnected = Some(error.to_string());
                    break;
                }
            }
        }

        if self.drop_upstream {
            self.drop_upstream = false;
            drop(upstream);
        } else if disconnected.is_none() {
            self.upstream = Some(upstream);
        }

        if let Some(message) = disconnected {
            eprintln!("[webd {}] upstream lost: {message}", self.id);
            self.core.reset_after_disconnect();
            self.pending_id = None;
            self.pending_since = None;
            if let Some(ws) = ws.as_mut() {
                send_json(
                    ws,
                    &ServerMessage::Error {
                        text: "Connection to the game server was lost",
                    },
                );
                self.send_view(ws);
            }
        }

        self.ws = ws;
    }

    fn check_request_timeout(&mut self) {
        let Some(sent) = self.pending_since else {
            return;
        };
        if sent.elapsed() < REQUEST_TIMEOUT {
            return;
        }

        let id = self.pending_id.take();
        self.pending_since = None;
        self.upstream = None;
        let mut ws = self.ws.take();
        if let Some(id) = id {
            let effects = self.core.handle_reply_id(id, "ERROR Request timed out");
            self.apply_effects(effects, ws.as_mut());
            if let Some(ws) = ws.as_mut() {
                send_json(
                    ws,
                    &ServerMessage::Notice {
                        text: "Request timed out. Disconnected.",
                    },
                );
            }
        }
        self.core.reset_after_disconnect();
        if let Some(ws) = ws.as_mut() {
            self.send_view(ws);
        }
        self.ws = ws;
    }

    fn apply_effects(&mut self, effects: Vec<Effect>, mut ws: Option<&mut WebSocket<TcpStream>>) {
        for effect in effects {
            match effect {
                Effect::Print(lines) => {
                    // A multi-line print is a terminal table; the structured
                    // `View` carries that data to the browser.
                    if lines.len() == 1 {
                        if let (Some(socket), Some(line)) = (ws.as_deref_mut(), lines.first()) {
                            send_json(socket, &ServerMessage::Notice { text: line });
                        }
                    }
                }
                Effect::SendRequest { id, payload } => {
                    self.pending_id = Some(id);
                    self.pending_since = Some(Instant::now());
                    match self.ensure_upstream() {
                        Ok(()) => {
                            if let Some(upstream) = self.upstream.as_mut() {
                                if let Err(error) = upstream.write_request(id, None, &payload) {
                                    let text = format!("Could not send request: {error}");
                                    if let Some(socket) = ws.as_deref_mut() {
                                        send_json(socket, &ServerMessage::Error { text: &text });
                                    }
                                    self.drop_upstream = true;
                                    self.upstream = None;
                                }
                            }
                        }
                        Err(text) => {
                            self.pending_id = None;
                            self.pending_since = None;
                            if let Some(socket) = ws.as_deref_mut() {
                                send_json(socket, &ServerMessage::Error { text: &text });
                            }
                        }
                    }
                }
                Effect::Disconnect => {
                    self.upstream = None;
                    self.drop_upstream = true;
                    self.core.reset_after_disconnect();
                }
                Effect::Render(_) | Effect::ClearRegion | Effect::Quit => {}
            }
        }

        if let Some(socket) = ws {
            self.send_view(socket);
        }
    }

    fn send_view(&self, ws: &mut WebSocket<TcpStream>) {
        let Some(view) = self.core.view() else {
            return;
        };
        send_json(ws, &ServerMessage::View { view: &view });
    }
}

fn send_message(ws: &mut WebSocket<TcpStream>, message: Message) -> Result<(), WsError> {
    // The read side is nonblocking; writes are small and should not fail
    // half-way because a browser is briefly not draining its socket.
    let _ = ws.get_ref().set_nonblocking(false);
    let result = ws.send(message);
    let _ = ws.get_ref().set_nonblocking(true);
    result
}

fn send_json(ws: &mut WebSocket<TcpStream>, message: &ServerMessage<'_>) {
    let Ok(text) = serde_json::to_string(message) else {
        return;
    };
    let _ = send_message(ws, Message::Text(text));
}

fn close_socket(ws: &mut WebSocket<TcpStream>) {
    let _ = ws.get_ref().set_nonblocking(false);
    let _ = ws.close(None);
    let _ = ws.get_ref().set_nonblocking(true);
}

#[cfg(test)]
mod tests {
    use super::*;
    use zagan_bet::client::view::{LobbyView, View};

    #[test]
    fn reads_only_well_formed_session_cookies() {
        let id = "a".repeat(64);
        assert_eq!(
            cookie_id(&format!("other=1; {COOKIE_NAME}={id}; theme=dark")),
            Some(id.clone())
        );
        assert_eq!(
            cookie_id(&format!("{HOST_COOKIE_NAME}={id}")),
            Some(id),
            "the secure __Host- name is accepted"
        );
        assert_eq!(cookie_id(&format!("{COOKIE_NAME}=short")), None);
        assert_eq!(cookie_id("other=1"), None);
        assert_eq!(cookie_id(""), None);
    }

    #[test]
    fn parses_a_browser_command_and_rejects_nonsense() {
        let message: ClientMessage = serde_json::from_str(
            r#"{"type":"command","command":{"Login":{"user":"alice","credential":"abc"}}}"#,
        )
        .expect("login command");
        match message {
            ClientMessage::Command { command } => {
                assert_eq!(
                    command,
                    Command::Login {
                        user: "alice".to_string(),
                        credential: "abc".to_string()
                    }
                );
            }
            _ => panic!("expected a command"),
        }

        assert!(serde_json::from_str::<ClientMessage>("{}").is_err());
        assert!(serde_json::from_str::<ClientMessage>(r#"{"type":"nope"}"#).is_err());
    }

    #[test]
    fn serializes_a_view_for_the_browser() {
        let view = View::Lobby(LobbyView {
            connected: true,
            user: Some("alice".to_string()),
            account: Some("1".to_string()),
            balance: Some(1000),
            games: vec![("slots".to_string(), "Slots".to_string())],
            login_error: None,
        });
        let json = serde_json::to_string(&ServerMessage::View { view: &view }).unwrap();
        assert!(json.contains(r#""type":"view""#));
        assert!(json.contains(r#""screen":"lobby""#));
        assert!(json.contains(r#""user":"alice""#));
    }
}