use std::io::{Read, Write};
use std::net::TcpStream;

use crossbeam_channel::{select, unbounded, Receiver, Sender};
use log::{debug, info, trace};

use crate::Packet;
use crate::{Error, Result};

/// AGW proxy stream.
pub struct Proxy {
    up: ConnectionV2,
    down: ConnectionV2,
}

impl Proxy {
    /// Set up a new proxy.
    ///
    /// # Errors
    ///
    /// If failing to connect to upstream.
    pub fn new(down: TcpStream) -> Result<Self> {
        let addr = "127.0.0.1:8010";
        let up = TcpStream::connect(addr)?;
        Ok(Self {
            up: ConnectionV2::new(up)?,
            down: ConnectionV2::new(down)?,
        })
    }
    /// Run the proxy forever.
    ///
    /// # Errors
    ///
    /// If the underlying connections fail, or other protocol error.
    pub fn run(
        &mut self,
        cb_up: &dyn Fn(Packet) -> Option<Packet>,
        cb_down: &dyn Fn(Packet) -> Option<Packet>,
    ) -> Result<()> {
        info!("Running proxy");
        self.up.send(Packet::VersionQuery)?;
        loop {
            select! {
                    recv(self.down.rx) -> packet => {
                        debug!("Got {packet:?} from downstream");
                match packet {
                Ok(packet) => {
                    if let Some(p) = cb_down(packet) {
                    debug!("… Transformed into {p:?}");
                    }
                },
                Err(_e) => return Ok(()),
                }
                    },
                    recv(self.up.rx) -> packet => {
                        debug!("Got {packet:?} from upstream");
                        match packet {
                Ok(packet) => {
                    if let Some(p) = cb_up(packet) {
                    debug!("Transformed into {p:?}");
                    }
                },
                Err(_e) => return Ok(()),
                        }
                    },
            };
        }
    }
}

struct ConnectionV2 {
    rx: Receiver<Packet>,
    tx: Sender<Packet>,
    rxthread: Option<std::thread::JoinHandle<Result<()>>>,
    txthread: Option<std::thread::JoinHandle<Result<()>>>,
}

impl Drop for ConnectionV2 {
    fn drop(&mut self) {
        debug!("Awaiting proxy thread shutdown");
        let _ = self.txthread.take().unwrap().join();
        let _ = self.rxthread.take().unwrap().join();
    }
}
impl ConnectionV2 {
    fn rx_loop(mut rstream: TcpStream, tx: &Sender<Packet>) -> Result<()> {
        loop {
            let mut header = [0_u8; crate::HEADER_LEN];
            rstream.read_exact(&mut header)?;

            let header = crate::parse_header(&header)?;
            let payload = if header.data_len > 0 {
                let mut payload = vec![0; header.data_len as usize];
                rstream.read_exact(&mut payload)?;
                payload
            } else {
                Vec::new()
            };
            //let reply = parse_reply(&header, &payload)?;
            //tx.send((header, reply))?;
            let packet = Packet::parse(&header, &payload)?;
            trace!("ConnectionV2 rx_loop: {packet:?}");
            tx.send(packet).map_err(Error::other)?;
        }
    }
    fn new(rstream: TcpStream) -> Result<Self> {
        let mut wstream = rstream.try_clone()?;
        let (rxtx, rxrx) = unbounded::<Packet>();
        let rxthread = std::thread::spawn(move || -> Result<()> { Self::rx_loop(rstream, &rxtx) });
        let (txtx, txrx) = unbounded::<Packet>();
        let txthread = std::thread::spawn(move || {
            for packet in txrx {
                let bytes = packet.serialize();
                let _ = wstream.write(&bytes)?;
                debug!("Send: {bytes:?}");
            }
            Ok(())
        });
        Ok(Self {
            rxthread: Some(rxthread),
            txthread: Some(txthread),
            rx: rxrx,
            tx: txtx,
        })
    }
    fn send(&self, packet: Packet) -> Result<()> {
        self.tx.send(packet).map_err(Error::other)
    }
}
