//
// shell.rs
//
// Copyright (C) 2022-2026 Posit Software, PBC. All rights reserved.
//
//

use amalthea::comm::comm_channel::Comm;
use amalthea::comm::comm_channel::CommMsg;
use amalthea::language::shell_handler::CommHandled;
use amalthea::language::shell_handler::ShellHandler;
use amalthea::socket::comm::CommSocket;
use amalthea::wire::complete_reply::CompleteReply;
use amalthea::wire::complete_request::CompleteRequest;
use amalthea::wire::execute_reply::ExecuteReply;
use amalthea::wire::execute_request::ExecuteRequest;
use amalthea::wire::history_reply::HistoryReply;
use amalthea::wire::history_request::HistoryRequest;
use amalthea::wire::inspect_reply::InspectReply;
use amalthea::wire::inspect_request::InspectRequest;
use amalthea::wire::is_complete_reply::IsComplete;
use amalthea::wire::is_complete_reply::IsCompleteReply;
use amalthea::wire::is_complete_request::IsCompleteRequest;
use amalthea::wire::jupyter_message::Status;
use amalthea::wire::kernel_info_reply::KernelInfoReply;
use amalthea::wire::kernel_info_request::KernelInfoRequest;
use amalthea::wire::language_info::LanguageInfo;
use amalthea::wire::language_info::LanguageInfoPositron;
use amalthea::wire::originator::Originator;
use async_trait::async_trait;
use bus::BusReader;
use crossbeam::channel::bounded;
use crossbeam::channel::unbounded;
use crossbeam::channel::Receiver;
use crossbeam::channel::Sender;
use harp::environment::R_ENVS;
use harp::line_ending::convert_line_endings;
use harp::line_ending::LineEnding;
use harp::object::RObject;
use harp::ParseResult;
use log::*;
use serde_json::json;

use crate::ark_comm::ArkComm;
use crate::console::KernelInfo;
use crate::console::SessionMode;
use crate::data_explorer::r_data_explorer::DATA_EXPLORER_COMM_NAME;
use crate::help::r_help::RHelp;
use crate::help::r_help::HELP_COMM_NAME;
use crate::plots::graphics_device::PLOT_COMM_NAME;
use crate::r_task;
use crate::request::KernelRequest;
use crate::request::RRequest;
use crate::ui::UiComm;
use crate::ui::UI_COMM_NAME;
use crate::variables::r_variables::RVariables;

pub struct Shell {
    r_request_tx: Sender<RRequest>,
    kernel_request_tx: Sender<KernelRequest>,
    kernel_init_rx: BusReader<KernelInfo>,
    kernel_info: Option<KernelInfo>,
}

#[derive(Debug)]
pub enum REvent {
    Prompt,
}

impl Shell {
    /// Creates a new instance of the shell message handler.
    pub(crate) fn new(
        r_request_tx: Sender<RRequest>,
        kernel_init_rx: BusReader<KernelInfo>,
        kernel_request_tx: Sender<KernelRequest>,
    ) -> Self {
        Self {
            r_request_tx,
            kernel_request_tx,
            kernel_init_rx,
            kernel_info: None,
        }
    }

    fn r_handle_is_complete_request(
        &self,
        req: &IsCompleteRequest,
    ) -> amalthea::Result<IsCompleteReply> {
        match harp::parse_status(&harp::ParseInput::Text(req.code.as_str())) {
            Ok(ParseResult::Complete(_)) => Ok(IsCompleteReply {
                status: IsComplete::Complete,
                indent: String::from(""),
            }),
            Ok(ParseResult::Incomplete) => Ok(IsCompleteReply {
                status: IsComplete::Incomplete,
                indent: String::from("+"),
            }),
            Err(_) | Ok(ParseResult::SyntaxError { .. }) => Ok(IsCompleteReply {
                status: IsComplete::Invalid,
                indent: String::from(""),
            }),
        }
    }
}

#[async_trait]
impl ShellHandler for Shell {
    async fn handle_info_request(
        &mut self,
        _req: &KernelInfoRequest,
    ) -> amalthea::Result<KernelInfoReply> {
        // Wait here for kernel initialization if it hasn't completed. This is
        // necessary for two reasons:
        //
        // 1. The kernel info response must include the startup banner, which is
        //    not emitted until R is done starting up.
        // 2. Jupyter frontends typically wait for the kernel info response to
        //    be sent before they signal that the kernel as ready for use, so
        //    blocking here ensures that it doesn't try to execute code before R is
        //    ready.
        if self.kernel_info.is_none() {
            trace!("Got kernel info request; waiting for R to complete initialization");
            self.kernel_info = Some(self.kernel_init_rx.recv().unwrap());
            trace!("R completed initialization, replying to kernel info request");
        } else {
            trace!("Got kernel info request; R has already started, replying to kernel info request with existing kernel information")
        }
        let kernel_info = self.kernel_info.as_ref().unwrap();

        let info = LanguageInfo {
            name: String::from("R"),
            version: kernel_info.version.clone(),
            file_extension: String::from(".R"),
            mimetype: String::from("text/r"),
            pygments_lexer: None,
            codemirror_mode: None,
            nbconvert_exporter: None,
            positron: Some(LanguageInfoPositron {
                input_prompt: kernel_info.input_prompt.clone(),
                continuation_prompt: kernel_info.continuation_prompt.clone(),
                build_version: Some(String::from(crate::BUILD_VERSION)),
                commit: Some(String::from(crate::BUILD_GIT_HASH)),
            }),
        };
        let mut supported_features = vec![String::from("debugger")];
        if matches!(kernel_info.session_mode, SessionMode::Notebook) {
            supported_features.push(String::from("proactive breakpoints"));
        }

        Ok(KernelInfoReply {
            status: Status::Ok,
            banner: kernel_info.banner.clone(),
            debugger: true,
            help_links: Vec::new(),
            language_info: info,
            implementation: String::from("ark"),
            // We use CARGO_PKG_VERSION here vs. the build version override since the Jupyter spec
            // specifically calls out the X.Y.Z format for this value
            implementation_version: String::from(env!("CARGO_PKG_VERSION")),
            supported_features,
        })
    }

    async fn handle_complete_request(
        &self,
        _req: &CompleteRequest,
    ) -> amalthea::Result<CompleteReply> {
        // No matches in this toy implementation.
        Ok(CompleteReply {
            matches: Vec::new(),
            status: Status::Ok,
            cursor_start: 0,
            cursor_end: 0,
            metadata: json!({}),
        })
    }

    /// Handle a request to test code for completion.
    async fn handle_is_complete_request(
        &self,
        req: &IsCompleteRequest,
    ) -> amalthea::Result<IsCompleteReply> {
        r_task(|| self.r_handle_is_complete_request(req))
    }

    /// Handles an ExecuteRequest by sending the code to the R execution thread
    /// for processing.
    fn start_execute_request(
        &mut self,
        originator: Originator,
        req: &ExecuteRequest,
    ) -> crossbeam::channel::Receiver<amalthea::Result<ExecuteReply>> {
        let (response_tx, response_rx) = unbounded::<amalthea::Result<ExecuteReply>>();
        let mut req_clone = req.clone();
        req_clone.code = convert_line_endings(&req_clone.code, LineEnding::Posix);
        if let Err(err) = self.r_request_tx.send(RRequest::ExecuteCode(
            req_clone.clone(),
            originator,
            response_tx,
        )) {
            warn!(
                "Could not deliver execution request to execution thread: {}",
                err
            )
        }

        trace!("Code sent to R: {}", req_clone.code);

        response_rx
    }

    /// Handles an introspection request
    async fn handle_inspect_request(&self, req: &InspectRequest) -> amalthea::Result<InspectReply> {
        let data = match req.code.as_str() {
            "err" => {
                json!({"text/plain": "This generates an error!"})
            },
            "teapot" => {
                json!({"text/plain": "This is clearly a teapot."})
            },
            _ => serde_json::Value::Null,
        };
        Ok(InspectReply {
            status: Status::Ok,
            found: data != serde_json::Value::Null,
            data,
            metadata: json!({}),
        })
    }

    async fn handle_history_request(
        &self,
        _req: &HistoryRequest,
    ) -> amalthea::Result<HistoryReply> {
        Ok(HistoryReply {
            status: Status::Ok,
            history: vec![],
        })
    }

    /// Handle a request to open a new comm channel
    ///
    /// Note that there might be multiple requests during a single session if
    /// the UI has been disconnected and reconnected.
    fn handle_comm_open(
        &mut self,
        target: Comm,
        comm: CommSocket,
        data: serde_json::Value,
    ) -> amalthea::Result<(bool, Option<Receiver<()>>)> {
        match target {
            Comm::Variables => Ok((handle_comm_open_variables(comm)?, None)),
            Comm::Ui => handle_comm_open_ui(comm, self.kernel_request_tx.clone(), data),
            Comm::Help => handle_comm_open_help(comm, self.kernel_request_tx.clone()),
            Comm::Other(target_name) if target_name == "ark" => {
                Ok((ArkComm::handle_comm_open(comm)?, None))
            },
            _ => Ok((false, None)),
        }
    }

    fn handle_comm_msg(
        &mut self,
        comm_id: &str,
        comm_name: &str,
        msg: CommMsg,
        originator: Originator,
    ) -> amalthea::Result<(CommHandled, Option<Receiver<()>>)> {
        match comm_name {
            DATA_EXPLORER_COMM_NAME | HELP_COMM_NAME | PLOT_COMM_NAME | UI_COMM_NAME => {
                let done_rx = self.start_kernel_request(|done_tx| KernelRequest::CommMsg {
                    comm_id: comm_id.to_string(),
                    msg,
                    originator: Box::new(originator),
                    done_tx,
                })?;
                Ok((CommHandled::Handled, Some(done_rx)))
            },
            _ => Ok((CommHandled::NotHandled, None)),
        }
    }

    fn handle_comm_close(
        &mut self,
        comm_id: &str,
        comm_name: &str,
    ) -> amalthea::Result<(CommHandled, Option<Receiver<()>>)> {
        match comm_name {
            DATA_EXPLORER_COMM_NAME | HELP_COMM_NAME | PLOT_COMM_NAME | UI_COMM_NAME => {
                let done_rx = self.start_kernel_request(|done_tx| KernelRequest::CommClose {
                    comm_id: comm_id.to_string(),
                    done_tx,
                })?;
                Ok((CommHandled::Handled, Some(done_rx)))
            },
            _ => Ok((CommHandled::NotHandled, None)),
        }
    }
}

impl Shell {
    /// Send a `KernelRequest` to the R thread and return a completion receiver.
    /// The caller (Shell) select-loops on this receiver and `comm_event_rx`
    /// to drain comm events while the request is processed.
    fn start_kernel_request(
        &self,
        build: impl FnOnce(Sender<()>) -> KernelRequest,
    ) -> amalthea::Result<Receiver<()>> {
        let (done_tx, done_rx) = bounded(0);
        self.kernel_request_tx
            .send(build(done_tx))
            .map_err(|err| amalthea::Error::SendError(err.to_string()))?;
        Ok(done_rx)
    }
}

fn handle_comm_open_variables(comm: CommSocket) -> amalthea::Result<bool> {
    r_task(|| {
        let global_env = RObject::view(R_ENVS.global);
        RVariables::start(global_env, comm);
        Ok(true)
    })
}

fn handle_comm_open_ui(
    comm: CommSocket,
    kernel_request_tx: Sender<KernelRequest>,
    data: serde_json::Value,
) -> amalthea::Result<(bool, Option<Receiver<()>>)> {
    let handler = UiComm::new(data);

    let (done_tx, done_rx) = bounded(0);
    kernel_request_tx
        .send(KernelRequest::CommOpen {
            comm_id: comm.comm_id.clone(),
            comm_name: comm.comm_name.clone(),
            outgoing_tx: comm.outgoing_tx.clone(),
            handler: Box::new(handler),
            done_tx,
        })
        .map_err(|err| amalthea::Error::SendError(err.to_string()))?;

    Ok((true, Some(done_rx)))
}

fn handle_comm_open_help(
    comm: CommSocket,
    kernel_request_tx: Sender<KernelRequest>,
) -> amalthea::Result<(bool, Option<Receiver<()>>)> {
    // Register the handler on the R thread, like the UI comm. `RHelp::handle_open`
    // starts the R help server and proxy and records their ports on `Console`, all
    // on the R thread. The handler holds the proxy's drop guard, so the proxy is
    // torn down when the comm is removed.
    let (done_tx, done_rx) = bounded(0);
    kernel_request_tx
        .send(KernelRequest::CommOpen {
            comm_id: comm.comm_id.clone(),
            comm_name: comm.comm_name.clone(),
            outgoing_tx: comm.outgoing_tx.clone(),
            handler: Box::new(RHelp::default()),
            done_tx,
        })
        .map_err(|err| amalthea::Error::SendError(err.to_string()))?;

    Ok((true, Some(done_rx)))
}
