use std::ops::Deref;
use std::sync::LazyLock;
use std::sync::Mutex;

use amalthea::comm::comm_channel::CommMsg;
use amalthea::comm::event::CommEvent;
use amalthea::comm_rpc_message;
use amalthea::socket::comm::CommInitiator;
use amalthea::socket::comm::CommOutgoingTx;
use amalthea::socket::comm::CommSocket;
use amalthea::socket::iopub::IOPubMessage;
use anyhow::Context;
use crossbeam::channel::Sender;
use harp::utils::r_is_null;
use harp::RObject;
use libr::R_NilValue;
use libr::SEXP;
use serde_json::json;
use stdext::result::ResultExt;
use stdext::spawn;
use stdext::unwrap;
use uuid::Uuid;

use crate::console::Console;

static RETICULATE_OUTGOING_TX: LazyLock<Mutex<Option<CommOutgoingTx>>> =
    LazyLock::new(|| Mutex::new(None));

// Each ark session has a unique session id which is used by the Positron frontend
// to associate a client id with a session.
static RETICULATE_ID: LazyLock<String> = LazyLock::new(|| format!("reticulate-{}", Uuid::new_v4()));

#[derive(Clone)]
pub struct ReticulateService {
    comm: CommSocket,
    comm_event_tx: Sender<CommEvent>,
}

impl ReticulateService {
    fn start(
        comm_id: String,
        comm_event_tx: Sender<CommEvent>,
        iopub_tx: Sender<IOPubMessage>,
        input: Option<String>,
    ) -> anyhow::Result<()> {
        let comm = CommSocket::new(
            CommInitiator::BackEnd,
            comm_id.clone(),
            String::from("positron.reticulate"),
            iopub_tx,
        );

        {
            let mut outgoing_tx_guard = RETICULATE_OUTGOING_TX.lock().unwrap();
            *outgoing_tx_guard = Some(comm.outgoing_tx.clone());
        }

        let service = Self {
            comm,
            comm_event_tx,
        };

        let event = CommEvent::Opened(
            service.comm.clone(),
            json!({
                "input": input,
                "reticulate_id": (*RETICULATE_ID).clone(),
            }),
            None,
        );
        service.comm_event_tx.send(event).log_err();

        spawn!(format!("ark-reticulate-{}", comm_id), move || {
            service
                .handle_messages()
                .context("Reticulate: Error handling messages")
                .log_err();
        });

        Ok(())
    }

    fn handle_messages(&self) -> Result<(), anyhow::Error> {
        loop {
            let msg: CommMsg = unwrap!(self.comm.incoming_rx.recv(), Err(err) => {
                log::error!("Reticulate: Error while receiving message from frontend: {err:?}");
                break;
            });

            log::trace!("Reticulate: Received message from front end: {msg:?}");

            if let CommMsg::Close = msg {
                break;
            }
        }

        // before finalizing the thread we make sure to send a close message to the front end
        self.comm.outgoing_tx.send(CommMsg::Close).log_err();

        // Reset the global soccket before closing
        log::info!("Reticulate Thread closing {:?}", self.comm.comm_id);
        {
            let mut outgoing_tx_guard = RETICULATE_OUTGOING_TX.lock().unwrap();
            *outgoing_tx_guard = None;
        }

        Ok(())
    }
}

// Creates a client instance reticulate can use to communicate with the front-end.
// We should aim at having at most **1** client per R session.
// Further actions that reticulate can ask the front-end can be requested through
// the comm_id that is returned by this function.
#[harp::register]
pub unsafe extern "C-unwind" fn ps_reticulate_open(input: SEXP) -> Result<SEXP, anyhow::Error> {
    let console = Console::get();

    let input: RObject = input.into();
    // Reticulate sends `NULL` or a string with the code to be executed in the Python console.
    let input_code: Option<String> = if r_is_null(input.sexp) {
        None
    } else {
        Some(input.try_into()?)
    };

    // If there's an id already registered, we just need to send the focus event
    {
        let outgoing_tx_guard = RETICULATE_OUTGOING_TX.lock().unwrap();
        if let Some(outgoing_tx) = outgoing_tx_guard.deref() {
            // There's a comm_id registered, we just send the focus event
            outgoing_tx.send(comm_rpc_message!("focus", input = input_code))?;
            return Ok(R_NilValue);
        }
    }

    let id = format!("reticulate-{}", Uuid::new_v4());
    ReticulateService::start(
        id,
        console.comm_event_tx().clone(),
        console.iopub_tx().clone(),
        input_code,
    )?;

    Ok(R_NilValue)
}

#[harp::register]
pub unsafe extern "C-unwind" fn ps_reticulate_id() -> Result<SEXP, anyhow::Error> {
    Ok(RObject::from((*RETICULATE_ID).clone()).sexp)
}
