use std::{
    collections::VecDeque,
    io::Error,
    net::{IpAddr, SocketAddr},
    pin::Pin,
    sync::{Arc, OnceLock, Weak},
    task::{Context, Poll},
};

use common::{
    crypto::MhyXorpad,
    net::{
        kcp_server::{KcpConnMgr, KcpSession},
        tools::Packet,
        ServiceType,
    },
    tools::time_utils,
    ServiceQualifier,
};
use mhy_kcp::Kcp;
use proto::{
    CmdID, ENetReason, GetPlayerTokenReq, GetPlayerTokenRsp, PacketHead, PingReq, PingRsp,
    PlayerLoginReq,
};
use tokio::{
    io::AsyncWrite,
    sync::{Mutex, OnceCell},
};

use crate::gateserver_service;

pub struct PlayerSession {
    session_id: u32,
    token: u32,
    recv_buf_len: usize,
    udp_remote_endpoint: SocketAddr,
    is_editor: bool,
    kcp: Mutex<Kcp<KcpOutput>>,
    connect_time: u64,
    secret_key: OnceCell<MhyXorpad>,
    uid: OnceCell<u32>,
    account_uid: OnceCell<String>,
    account_token: OnceCell<String>,
    platform_type: OnceCell<u32>,
    login_finished: OnceCell<()>,
    recv_login: OnceCell<()>,
    gameserver_id: OnceCell<u32>,
}

impl PlayerSession {
    pub fn session_time(&self) -> u32 {
        (time_utils::get_now() - self.connect_time) as u32
    }

    async fn ikcp_input(&self, buf: &[u8]) -> Result<(), mhy_kcp::Error> {
        let mut kcp = self.kcp.lock().await;

        kcp.input(buf)?;
        kcp.async_update(self.session_time()).await?;
        kcp.async_flush().await
    }

    async fn on_recv(&self, data: &mut [u8]) -> Result<(), Box<dyn std::error::Error>> {
        self.encrypt_packet_data(data).await;
        let mut packet = match Packet::parse_from_array(data) {
            Ok(packet) => packet,
            Err(err) => {
                tracing::warn!(
                    "Packet::parse_from_array failed, uid: {}, session_id: {}, err: {}",
                    self.get_uid(),
                    self.session_id,
                    err
                );
                return Ok(());
            }
        };

        self.init_packet_head(&mut packet);
        packet.head.enet_is_reliable = 1;

        if !self.is_finish_login() && packet.cmd_id == PingReq::CMD_ID {
            let Ok(req) = packet.get_proto::<PingReq>() else {
                tracing::warn!(
                    "get_proto PingReq failed, session_id={}, ip:{}, port:{}",
                    self.session_id,
                    self.udp_remote_endpoint.ip(),
                    self.udp_remote_endpoint.port()
                );
                return Ok(());
            };

            let rsp = PingRsp {
                retcode: 0,
                client_time: req.client_time,
                seq: req.seq,
            };

            let mut packet = Packet::new(rsp, PacketHead::default());

            self.init_packet_head(&mut packet);
            tracing::info!(
                "recv ping before login finish, session_id={} uid:{} ip:{} port:{}",
                self.session_id,
                self.get_uid(),
                self.udp_remote_endpoint.ip(),
                self.udp_remote_endpoint.port()
            );

            return self.send_packet(packet).await;
        }

        if self.uid.get().is_none() {
            if packet.cmd_id != GetPlayerTokenReq::CMD_ID && packet.cmd_id != PingReq::CMD_ID {
                tracing::error!(
                    "refuse packet cmd_id={} because uid=0, ip:{}, port:{}",
                    packet.cmd_id,
                    self.udp_remote_endpoint.ip(),
                    self.udp_remote_endpoint.port()
                );

                gateserver_service()
                    .disconnect(self.session_id, ENetReason::EnetLoginUnfinished)
                    .await;
            } else if packet.cmd_id == GetPlayerTokenReq::CMD_ID {
                self.on_get_player_token_req(packet).await;
            }
        } else {
            if !self.is_recv_login() && packet.cmd_id != PlayerLoginReq::CMD_ID {
                tracing::error!("need LoginReq first cmd_id: {}", packet.cmd_id);
                gateserver_service()
                    .disconnect(self.session_id, ENetReason::EnetLoginUnfinished)
                    .await;

                return Ok(());
            }

            if packet.cmd_id == PlayerLoginReq::CMD_ID {
                let _ = self.recv_login.set(());
                gateserver_service()
                    .login_handler
                    .on_player_login_req(packet)
                    .await;
            } else if self.is_recv_login() {
                gateserver_service()
                    .service_box
                    .push_packet_to_service(
                        ServiceQualifier::AppId(self.get_gameserver_id()),
                        packet,
                    )
                    .await?;
            }
        }

        Ok(())
    }

    pub async fn send_packet(&self, packet: Packet) -> Result<(), Box<dyn std::error::Error>> {
        let mut buf = packet.to_vec();

        if packet.cmd_id == GetPlayerTokenRsp::CMD_ID {
            self.encrypt_first_packet_data(&mut buf).await;
        } else {
            self.encrypt_packet_data(&mut buf).await;
        }

        let mut kcp = self.kcp.lock().await;
        kcp.send(&buf)?;
        kcp.async_flush().await?;
        kcp.async_update(self.session_time()).await?;

        Ok(())
    }

    fn init_packet_head(&self, packet: &mut Packet) {
        let head = &mut packet.head;
        head.user_id = self.get_uid();
        head.user_session_id = self.session_id;
        head.user_ip = self.get_remote_ip();
        head.source_service = ServiceType::Gateserver as u32;
        head.sender_app_id = gateserver_service().config.app_id;

        // TODO: tothemoon and pathfinding
        head.target_service = ServiceType::Gameserver as u32;
        head.rpc_id = self.get_gameserver_id();
    }

    pub fn get_uid(&self) -> u32 {
        self.uid.get().map(|u| *u).unwrap_or_default()
    }

    pub fn set_uid(&self, uid: u32) {
        let _ = self.uid.set(uid);
    }

    pub fn get_account_token(&self) -> String {
        self.account_token.get().cloned().unwrap_or_default()
    }

    pub fn get_platform_type(&self) -> u32 {
        self.platform_type.get().cloned().unwrap_or_default()
    }

    pub fn set_gameserver_id(&self, id: u32) {
        let _ = self.gameserver_id.set(id);
    }

    pub fn get_gameserver_id(&self) -> u32 {
        self.gameserver_id.get().cloned().unwrap_or_default()
    }

    async fn on_get_player_token_req(&self, packet: Packet) {
        let Ok(get_player_token) = packet.get_proto::<GetPlayerTokenReq>() else {
            tracing::warn!(
                "get_proto GetPlayerToken failed, session_id:{}",
                self.session_id
            );
            return;
        };
        if self.account_uid.get().is_none() {
            self.account_uid
                .set(get_player_token.account_uid.clone())
                .unwrap();
            self.account_token
                .set(get_player_token.account_token.clone())
                .unwrap();
            self.platform_type
                .set(get_player_token.platform_type)
                .unwrap();
        } else {
            gateserver_service()
                .disconnect(self.session_id, ENetReason::EnetLoginUnfinished)
                .await;
            tracing::info!("[LOGIN] disconnect session_id:{} reason:{} cmd: req account_uid:{} session account_uid:{}", self.session_id, ENetReason::EnetLoginUnfinished.as_str_name(), get_player_token.account_uid, self.account_uid.get().unwrap());
            return;
        }

        tracing::info!(
            "[LOGIN] PlayerGetToken account_uid:{}",
            &get_player_token.account_uid
        );

        gateserver_service()
            .login_handler
            .on_get_player_token_req(packet)
            .await;
    }

    async fn encrypt_packet_data(&self, data: &mut [u8]) {
        if let Some(xorpad) = self.secret_key.get() {
            xorpad.xor(data);
        } else {
            self.encrypt_first_packet_data(data).await;
        }
    }

    async fn encrypt_first_packet_data(&self, data: &mut [u8]) {
        static FIRST_PACKET_SECRET_KEY: OnceLock<MhyXorpad> = OnceLock::new();

        let xorpad = match FIRST_PACKET_SECRET_KEY.get() {
            Some(xorpad) => xorpad,
            None => {
                let service = gateserver_service();
                let region_config = service
                    .deploy_config_mgr
                    .find_db_region_config(service.config.region.name.clone())
                    .await
                    .expect("Failed to find region config");
                FIRST_PACKET_SECRET_KEY
                    .get_or_init(|| MhyXorpad::from_ec2b(&region_config.client_secret_key))
            }
        };

        xorpad.xor(data);
    }

    pub fn set_secret_key(&self, secret_key: MhyXorpad) {
        let _ = self.secret_key.set(secret_key);
    }

    pub fn set_is_finish_login(&self) {
        let _ = self.login_finished.set(());
    }

    pub fn set_is_recv_login(&self) {
        let _ = self.recv_login.set(());
    }

    pub fn is_finish_login(&self) -> bool {
        self.login_finished.get().is_some()
    }

    pub fn is_recv_login(&self) -> bool {
        self.recv_login.get().is_some()
    }
}

struct KcpOutput {
    conn_mgr: Weak<KcpConnMgr<PlayerSession>>,
    udp_remote_endpoint: SocketAddr,
}

impl KcpSession for PlayerSession {
    fn new(
        conn_mgr: Arc<KcpConnMgr<Self>>,
        session_id: u32,
        token: u32,
        recv_buf_len: usize,
        udp_remote_endpoint: SocketAddr,
    ) -> Self
    where
        Self: Sized,
    {
        Self {
            session_id,
            token,
            recv_buf_len,
            udp_remote_endpoint,
            is_editor: false,
            kcp: Mutex::new(Kcp::new(
                session_id,
                token,
                false,
                KcpOutput {
                    conn_mgr: Arc::downgrade(&conn_mgr),
                    udp_remote_endpoint,
                },
            )),
            connect_time: time_utils::get_now(),
            secret_key: OnceCell::new(),
            uid: OnceCell::new(),
            account_uid: OnceCell::new(),
            account_token: OnceCell::new(),
            platform_type: OnceCell::new(),
            login_finished: OnceCell::new(),
            recv_login: OnceCell::new(),
            gameserver_id: OnceCell::new(),
        }
    }

    fn on_connect(&mut self, data: u32) {
        match ENetReason::try_from(data as i32) {
            Ok(ENetReason::EnetClientEditorConnectKey) => {
                tracing::info!(
                    "[LOGIN] player connect from unity editor, ip:{}, port:{}, session_id:{}",
                    self.udp_remote_endpoint.ip(),
                    self.udp_remote_endpoint.port(),
                    self.session_id
                );
                self.is_editor = true;
            }
            Ok(ENetReason::EnetClientConnectKey) => tracing::info!(
                "[LOGIN] player connect from mobile phone, ip:{}, port:{}, session_id:{}",
                self.udp_remote_endpoint.ip(),
                self.udp_remote_endpoint.port(),
                self.session_id
            ),
            _ => tracing::info!(
                "[LOGIN] invalid connect data:{}, ip:{}, port:{}, session_id:{}",
                data,
                self.udp_remote_endpoint.ip(),
                self.udp_remote_endpoint.port(),
                self.session_id
            ),
        }
    }

    async fn input(self: Arc<Self>, udp_data: Box<[u8]>) {
        if let Err(err) = self.ikcp_input(&udp_data).await {
            tracing::warn!(
                "ikcp_input failed, not a valid kcp packet, session_id={}, ip={}, port={}, bytes_recvd={}, ret={}",
                self.session_id,
                self.udp_remote_endpoint.ip(),
                self.udp_remote_endpoint.port(),
                udp_data.len(),
                err
            );
            return;
        }

        let mut queue = VecDeque::<Box<[_]>>::new();
        let mut buf = vec![0; self.recv_buf_len];
        while let Ok(len) = self.kcp.lock().await.recv(&mut buf) {
            queue.push_back(buf[..len].into());
        }

        while let Some(mut packet) = queue.pop_front() {
            if let Err(err) = self.on_recv(&mut packet).await {
                tracing::warn!("on_recv fail: {err}")
            }
        }
    }

    fn get_remote_ip(&self) -> u32 {
        match self.udp_remote_endpoint.ip() {
            IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()),
            IpAddr::V6(_) => panic!("ipv6 is not supported in this context"),
        }
    }

    fn get_user_session_id(&self) -> u32 {
        self.session_id
    }

    fn on_disconnect(&self, reason: u32) {
        // TODO: PlayerDisconnectNotify

        let reason = if let Ok(reason) = ENetReason::try_from(reason as i32) {
            reason.as_str_name().to_string()
        } else {
            reason.to_string()
        };

        tracing::info!("[LOGIN] player disconnect, uid:{}, connect time:{}, is editor:{}, reason:{}, session_id:{}",
        self.get_uid(), self.session_time() / 1000, self.is_editor, reason, self.session_id);
    }

    fn get_user_token(&self) -> u32 {
        self.token
    }

    fn get_udp_remote_endpoint(&self) -> SocketAddr {
        self.udp_remote_endpoint
    }
}

impl AsyncWrite for KcpOutput {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, Error>> {
        self.conn_mgr
            .upgrade()
            .unwrap()
            .send_udp_packet(cx, buf, self.udp_remote_endpoint)
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
        Poll::Ready(Ok(()))
    }
}
