ctoolbox/

workspace.rs

use crate::cli::{self, Cli};
use crate::storage::get_storage_dir;
use crate::utilities::ipc::{Channel, channel_from_args_and_key};
use crate::utilities::logging::setup_logger;
use crate::utilities::process::{Process, ProcessManager, start_process};
use crate::utilities::resource_lock::check_filesystem_lock_support;
use crate::utilities::*;
use crate::workspace::ipc_old::server::start_ipc_server;
use crate::workspace::ipc_old::subprocess_init;
use portpicker::pick_unused_port;
pub use serde_json::json as utilities_serde_json_json;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use text_io::read;

pub mod ipc;
pub mod ipc_old;
use once_cell::sync::OnceCell;

// FIXME: The IPC logic and API is confused. The workspace should be able to
// start any subprocess. Renderer processes should each also be able to start
// subprocesses. Other subprocesses should NOT be able to start their own
// subprocesses, and instead should call either their parent workspace or
// renderer when they need to make calls to other tools.

// Some services should be single instance processes: e.g. the workspace itself,
// storage manager, local web server, etc.
// Other services should be multi-instance, e.g. renderer processes, format
// converters, etc.

static PROCESS_MANAGER: OnceCell<Arc<Mutex<ProcessManager>>> = OnceCell::new();

/// Initializes the global process manager singleton.
pub fn set_global_process_manager(
    pm: Arc<Mutex<ProcessManager>>,
) -> anyhow::Result<()> {
    #[cfg(test)]
    {
        // In tests, allow re-initialization.
        let _ = PROCESS_MANAGER.get();
        if PROCESS_MANAGER.get().is_some() {
            return Ok(());
        }
    }
    PROCESS_MANAGER
        .set(pm)
        .map_err(|_| anyhow::anyhow!("ProcessManager already set"))
}

/// Gets a reference to the global process manager.
pub fn get_global_process_manager() -> anyhow::Result<Arc<Mutex<ProcessManager>>>
{
    PROCESS_MANAGER
        .get()
        .cloned()
        .ok_or_else(|| anyhow::anyhow!("ProcessManager not initialized"))
}

pub struct WorkspaceState {
    pub processes: Arc<Mutex<ProcessManager>>,
}

/// Initializes the process manager singleton and returns the `WorkspaceState`.
/// For the main workspace, this  knows about ALL the subprocesses. For each
/// subprocess invocation, it only knows about the workspace process, which then
/// delegates to other processes.
pub fn initialize_process_state(
    invocation: &cli::Invocation,
) -> anyhow::Result<WorkspaceState> {
    let mut pm;
    if let cli::Invocation::Subprocess(subproc) = &invocation {
        let authentication_key: String = read!("{}\n");
        let control_channel =
            channel_from_args_and_key(&subproc.ipc, authentication_key);
        pm = ProcessManager {
            port: 0,
            ipc_control_channel: control_channel.clone(),
            processes: HashMap::new(),
            ..Default::default()
        };
        pm.processes.insert(
            0,
            Process {
                pid: 0,
                system_pid: 0,
                channel: control_channel,
                parent_pid: 0,
            },
        );
    } else {
        pm = ProcessManager {
            port: 0,
            ipc_control_channel: Channel {
                name: String::new(),
                port: 0,
                authentication_key: String::new(),
            },
            processes: HashMap::new(),
            ..Default::default()
        };
        pm.processes.insert(
            0,
            Process {
                pid: 0,
                system_pid: 0,
                channel: Channel {
                    name: String::new(),
                    port: 0,
                    authentication_key: String::new(),
                },
                parent_pid: 0,
            },
        );

        setup_panic_hooks(&pm);
    }

    let pm_arc = Arc::new(Mutex::new(pm));
    set_global_process_manager(pm_arc.clone())?;

    Ok(WorkspaceState {
        processes: get_global_process_manager()?,
    })
}

pub async fn entry() -> anyhow::Result<()> {
    check_filesystem_lock_support()?;

    // Parse either a subprocess invocation or a user CLI command set.
    let invocation = cli::parse_invocation()?;

    // If it's a subprocess, initialize and exit early.
    if let cli::Invocation::Subprocess(_) = &invocation {
        setup_logger(&invocation)?;
        subprocess_init(&invocation)?;
        return Ok(());
    }
    // At this point we know it's a user invocation.
    let cli = invocation.expect_cli();

    // Try lightweight tools (hex2dec, etc.).
    if let Some(code) = cli::maybe_run_lightweight(cli).await? {
        // Allow tests to inspect code by returning an error code if desired,
        // but typically just exit here:
        std::process::exit(code);
    }

    // Proceed with full application boot.
    setup_logger(&invocation)?;
    let state = initialize_process_state(&invocation)?;
    boot(state, cli).await?;
    Ok(())
}

/// Sets up the IPC control channel and starts the IPC server.
/// Returns the IPC control channel and authentication key.
fn setup_ipc_server(
    pm_arc: &Arc<Mutex<ProcessManager>>,
    args: &Cli,
) -> anyhow::Result<(Channel, String)> {
    let port: u16 = if args.ctoolbox_ipc_port.is_none() {
        pick_unused_port().expect("No ports free")
    } else {
        args.ctoolbox_ipc_port.unwrap()
    };
    let authentication_key: String = generate_authentication_key();
    let ipc_control_channel: Channel = Channel {
        name: "ipc_control".to_string(),
        port,
        authentication_key: authentication_key.clone(),
    };

    let process_manager = pm_arc.lock();
    if process_manager.is_err() {
        return Err(anyhow::anyhow!("Failed to lock process manager"));
    }
    let mut process_manager = process_manager.unwrap();
    process_manager.port = port;
    process_manager.ipc_control_channel = ipc_control_channel.clone();
    drop(process_manager);

    let authentication_key_for_server = authentication_key.clone();
    log!("Waiting for IPC server to come up...");
    let pm_for_server = pm_arc.clone();
    std::thread::spawn(move || {
        start_ipc_server(pm_for_server, port, authentication_key_for_server);
    });
    loop {
        let response = maybe_send_message(
            &ipc_control_channel,
            json!({ "type": "ipc_ping" }).as_str(),
        );
        if response.is_ok() {
            break;
        }
        sleep(1);
    }
    log!("IPC server OK.");

    Ok((ipc_control_channel, authentication_key))
}

/// Main boot logic for normal application startup.
async fn boot(state: WorkspaceState, args: &Cli) -> Result<()> {
    let pm_arc = get_global_process_manager()?;
    let (ipc_control_channel, _authentication_key) =
        setup_ipc_server(&pm_arc, args)?;

    get_storage_dir().unwrap();

    let renderer =
        start_process(&pm_arc, 0, vec!["renderer".to_string()]).await?;

    let localwebui = start_process(&pm_arc, 0, vec!["io".to_string()]).await?;

    call_ipc(&localwebui.channel, "start_local_web_ui", "".into()).await?;
    /*
    print(strtovec(
        &call_ipc(&renderer.channel, "test_echo", "Hello, world!".into())
            .await?,
    ));

    sleep(1);

    let doc =
        storage::get_asset("intro.html").expect("Could not load intro.html");
    // let doc = strtovec("0");
    let doc_str = vectostr(&doc);
    print(strtovec(format!("Document: {doc_str}").as_str()));*/
    // let pid = hc_renderer::start(hc_formats::convert_from(doc, strtovec("html")));
    // hc_renderer::start(hc_formats::convert_from(doc, strtovec("html")));*/
    // data_channel_test().await;
    loop {
        sleep(10);
    }
}

pub fn workspace_restart(manager: &Arc<Mutex<ProcessManager>>) {
    std::process::exit(0);
}

pub fn workspace_shutdown(manager: &Arc<Mutex<ProcessManager>>) {
    let answer: String = read!("{}\n");
    if answer.to_lowercase().starts_with('y') {
        std::process::exit(0);
    }
}

/// Boot logic for tests: sets up IPC service and workspace process only.
fn impl_boot_for_test() -> Result<Arc<Mutex<ProcessManager>>> {
    setup_logger(&cli::Invocation::User(Cli {
        ctoolbox_ipc_port: None,
        command: None,
    }))?;

    // Setup a test-specific ProcessManager
    let mut pm = ProcessManager {
        port: 0,
        ipc_control_channel: Channel {
            name: String::new(),
            port: 0,
            authentication_key: String::new(),
        },
        processes: HashMap::new(),
        ..Default::default()
    };
    pm.processes.insert(
        0,
        Process {
            pid: 0,
            system_pid: 0,
            channel: Channel {
                name: String::new(),
                port: 0,
                authentication_key: String::new(),
            },
            parent_pid: 0,
        },
    );
    setup_panic_hooks(&pm);
    let pm_arc = Arc::new(Mutex::new(pm));

    // Stub: In tests, we do not start subprocesses or spawn threads.
    // Optionally, you could simulate IPC server startup here if needed.

    set_global_process_manager(pm_arc.clone())?;

    Ok(pm_arc)
}

pub struct TestHarness {
    pub pm: Arc<Mutex<ProcessManager>>,
}

static HARNESS: OnceCell<TestHarness> = OnceCell::new();

impl TestHarness {
    #[tracing::instrument]
    /// Attempts to initialize the `TestHarness`, returning a result.
    pub fn try_init() -> Result<&'static TestHarness> {
        // Only initialize if not already.
        if let Some(harness) = HARNESS.get() {
            return Ok(harness);
        }

        let pm = impl_boot_for_test();
        match pm {
            Ok(pm_arc) => {
                Ok(HARNESS.get_or_init(|| TestHarness { pm: pm_arc }))
            }
            Err(e) => Err(e),
        }
    }
}

/* impl Drop for TestHarness {
    fn drop(&mut self) {
        let pm = self.pm.lock().unwrap();
        cleanup_processes(&pm);
    }
} */

// Not meant to be used directly - use #[crate::ctb_test] instead.
pub fn boot_for_test() -> Option<&'static TestHarness> {
    if let Ok(harness) = TestHarness::try_init() {
        Some(harness)
    } else {
        // backtrace_print();
        None
    }
}