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

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;

use amalthea::comm::event::CommEvent;
use amalthea::connection_file::ConnectionFile;
use amalthea::kernel;
use amalthea::language::server_handler::ServerHandler;
use amalthea::registration_file::RegistrationFile;
use amalthea::socket::iopub::IOPubMessage;
use amalthea::socket::stdin::StdInRequest;
use bus::Bus;
use crossbeam::channel::bounded;
use crossbeam::channel::unbounded;

use crate::console::ConsoleNotification;
use crate::console::SessionMode;
use crate::control::Control;
use crate::dap;
use crate::lsp;
use crate::repos::DefaultRepos;
use crate::request::KernelRequest;
use crate::request::RRequest;
use crate::shell::Shell;

/// Exported for unit tests.
pub fn start_kernel(
    connection_file: ConnectionFile,
    registration_file: Option<RegistrationFile>,
    r_args: Vec<String>,
    startup_file: Option<String>,
    session_mode: SessionMode,
    capture_streams: bool,
    default_repos: DefaultRepos,
) {
    // Locate R home directory
    let r_home = match harp::command::r_home_setup() {
        Ok(r_home) => r_home,
        Err(err) => panic!("Can't set up `R_HOME`: {err}"),
    };

    let r_version = match crate::version::from_r_home(&r_home) {
        Ok(r_version) => r_version,
        Err(err) => panic!("Can't detect R version: {err:?}"),
    };

    if !r_version.is_supported() {
        panic!(
            "Unsupported R version {}.{}.{}. Ark requires R >= {}.{}.{}",
            r_version.major,
            r_version.minor,
            r_version.patch,
            crate::version::MIN_R_MAJOR,
            crate::version::MIN_R_MINOR,
            crate::version::MIN_R_PATCH,
        );
    };

    // Create the channels used for communication. These are created here
    // as they need to be shared across different components / threads.
    let (iopub_tx, iopub_rx) = bounded::<IOPubMessage>(10);

    // Create the pair of channels that will be used to relay messages from
    // the open comms
    let (comm_event_tx, comm_event_rx) = bounded::<CommEvent>(10);

    // A broadcast channel (bus) used to notify clients when the kernel
    // has finished initialization.
    let mut kernel_init_tx = Bus::new(1);

    // A channel pair used for kernel requests.
    // These events are used to manage the runtime state, and also to
    // handle message delivery, among other things.
    let (r_request_tx, r_request_rx) = bounded::<RRequest>(1);
    let (kernel_request_tx, kernel_request_rx) = bounded::<KernelRequest>(1);

    // Async communication channel with the R thread (Console)
    let (console_notification_tx, console_notification_rx) =
        tokio::sync::mpsc::unbounded_channel::<ConsoleNotification>();

    // Create the LSP and DAP clients.
    // Not all Amalthea kernels provide these, but ark does.
    // They must be able to deliver messages to the shell channel directly.
    let lsp = Arc::new(Mutex::new(lsp::handler::Lsp::new(
        r_home.clone(),
        kernel_init_tx.add_rx(),
        console_notification_tx.clone(),
    )));

    // DAP needs the `RRequest` channel to communicate with
    // `read_console()` and send commands to the debug interpreter
    let dap = dap::Dap::new_shared(r_request_tx.clone());

    // Communication channel between the R main thread and the Amalthea
    // StdIn socket thread
    let (stdin_request_tx, stdin_request_rx) = bounded::<StdInRequest>(1);

    // Create the shell.
    let kernel_init_rx = kernel_init_tx.add_rx();
    let shell = Box::new(Shell::new(
        r_request_tx.clone(),
        kernel_init_rx,
        kernel_request_tx,
    ));

    // Create the control handler; this is used to handle shutdown/interrupt and
    // related requests
    let control = Box::new(Control::new(
        r_request_tx.clone(),
        dap.clone(),
        iopub_tx.clone(),
        session_mode,
    ));

    // Create the stream behavior; this determines whether the kernel should
    // capture stdout/stderr and send them to the frontend as IOPub messages
    let stream_behavior = match capture_streams {
        true => amalthea::kernel::StreamBehavior::Capture,
        false => amalthea::kernel::StreamBehavior::None,
    };

    let (stdin_reply_tx, stdin_reply_rx) = unbounded();

    // Create the list of server handlers that Ark supports. Amalthea handles
    // the handshake part on startup.
    let mut server_handlers: HashMap<String, Arc<Mutex<dyn ServerHandler>>> = HashMap::new();
    server_handlers.insert("lsp".to_string(), lsp);
    server_handlers.insert("ark_dap".to_string(), dap.clone());

    let res = kernel::connect(
        "ark",
        connection_file,
        registration_file,
        kernel::Handlers {
            shell_handler: shell,
            control_handler: control,
            server_handlers,
        },
        kernel::ConnectionChannels {
            iopub_tx: iopub_tx.clone(),
            iopub_rx,
            comm_event_rx,
            stdin_request_rx,
            stdin_reply_tx,
        },
        stream_behavior,
    );
    if let Err(err) = res {
        panic!("Couldn't connect to frontend: {err:?}");
    }

    // Start parent process monitoring for graceful shutdown if applicable. Currently we
    // only do this for Linux since it uses `prctl()`.
    if let Err(err) = crate::sys::parent_monitor::start_parent_monitoring(r_request_tx.clone()) {
        log::error!("Failed to start parent process monitoring: {err}");
    }

    // Start R
    crate::console::Console::start(
        r_home,
        r_args,
        startup_file,
        comm_event_tx,
        r_request_rx,
        stdin_request_tx,
        stdin_reply_rx,
        iopub_tx,
        kernel_init_tx,
        kernel_request_rx,
        dap,
        session_mode,
        default_repos,
        console_notification_rx,
    )
}
