unifier/cli/

mod.rs

//! CLI dispatch.

mod defs;

use clap::Parser;
use defs::{
    ChrootCommands, Cli, Commands, DaemonCommands, LogCommands, NamespaceCommands, SqlCommands,
    TickCommands, TripleCommands, WebCommands,
};

use crate::chroot::{init_chroot, list_chroots};
use crate::daemon::{self};
use crate::home::UnifierHome;
use crate::log as spanlog;
use crate::namespace;
use crate::postbox::{
    ack, delete_key, get_key, list_dir, poll_cron, poll_mailbox, post_cron, put_key, send_from,
    Message,
};
use crate::sqlite::{self, SqlOutcome};
use crate::Result;

#[cfg(unix)]
use crate::daemon::{
    response_found, response_messages, response_ok, response_uuid, response_value, Client, Request,
};

pub fn run() -> Result<()> {
    let cli = Cli::parse();
    let home = UnifierHome::resolve(cli.home, cli.chroot)?;
    let ns = cli.namespace;

    match cli.cmd {
        Commands::Daemon(DaemonCommands::Start) => {
            daemon::start(&home, false)?;
            println!("daemon started");
            Ok(())
        }
        Commands::Daemon(DaemonCommands::Run) => {
            #[cfg(unix)]
            {
                daemon::run_server(home)
            }
            #[cfg(not(unix))]
            {
                Err(crate::Error::msg("hot daemon requires a Unix platform"))
            }
        }
        Commands::Daemon(DaemonCommands::Stop) => {
            daemon::stop(&home)?;
            println!("daemon stopped");
            Ok(())
        }
        Commands::Daemon(DaemonCommands::Status) => daemon::status(&home),
        Commands::Daemon(DaemonCommands::Flush) => daemon::flush(&home),
        Commands::Daemon(DaemonCommands::Watch) => {
            #[cfg(unix)]
            {
                daemon::ensure_running(&home)?;
                daemon::watch(&home)
            }
            #[cfg(not(unix))]
            {
                Err(crate::Error::msg("hot daemon requires a Unix platform"))
            }
        }
        Commands::Daemon(DaemonCommands::Gc { dry_run }) => {
            #[cfg(unix)]
            {
                let _ = home;
                daemon::gc(dry_run)
            }
            #[cfg(not(unix))]
            {
                let _ = (home, dry_run);
                Err(crate::Error::msg("hot daemon requires a Unix platform"))
            }
        }
        Commands::Chroot(ChrootCommands::Init { name }) => {
            init_chroot(home.global_path(), &name)?;
            println!(
                "chroot initialized: {}",
                home.global_path().join("chroots").join(name).display()
            );
            Ok(())
        }
        Commands::Chroot(ChrootCommands::List) => {
            for name in list_chroots(home.global_path())? {
                println!("{name}");
            }
            Ok(())
        }
        Commands::Namespace(cmd) => {
            home.ensure()?;
            dispatch_namespace(&home, cmd, ns.as_deref())
        }
        Commands::Sql(cmd) => {
            home.ensure()?;
            dispatch_sql(&home, cmd)
        }
        Commands::Triple(cmd) => {
            home.ensure()?;
            dispatch_triple(&home, cmd)
        }
        Commands::Log(cmd) => {
            home.ensure()?;
            dispatch_log(&home, cmd)
        }
        Commands::Serve {
            name,
            file,
            content_type,
            ttl,
            wrap,
            title,
        } => {
            home.ensure()?;
            dispatch_serve(
                &home,
                ServeOpts {
                    name,
                    file,
                    content_type,
                    ttl,
                    wrap,
                    title,
                    no_daemon: cli.no_daemon,
                },
            )
        }
        Commands::Web(cmd) => {
            home.ensure()?;
            dispatch_web(&home, cmd, cli.no_daemon)
        }
        cmd => {
            home.ensure()?;
            if cli.no_daemon {
                dispatch_data(&home, cmd, ns.as_deref())
            } else {
                #[cfg(unix)]
                {
                    daemon::ensure_running(&home)?;
                    dispatch_via_daemon(&home, cmd, ns.as_deref())
                }
                #[cfg(not(unix))]
                {
                    dispatch_data(&home, cmd, ns.as_deref())
                }
            }
        }
    }
}

fn dispatch_namespace(
    home: &UnifierHome,
    cmd: NamespaceCommands,
    override_ns: Option<&str>,
) -> Result<()> {
    match cmd {
        NamespaceCommands::Set { name } => {
            namespace::set(home, &name)?;
            println!("{name}");
            Ok(())
        }
        NamespaceCommands::Get => match namespace::effective(home, override_ns)? {
            Some(name) => {
                println!("{name}");
                Ok(())
            }
            None => Err(crate::Error::msg("no namespace")),
        },
        NamespaceCommands::Clear => {
            if namespace::clear(home)? {
                Ok(())
            } else {
                Err(crate::Error::msg("no namespace"))
            }
        }
    }
}

fn qualify(home: &UnifierHome, ns: Option<&str>, key: String) -> Result<String> {
    namespace::qualify_key(home, ns, &key)
}

fn dispatch_sql(home: &UnifierHome, cmd: SqlCommands) -> Result<()> {
    match cmd {
        SqlCommands::List => {
            for name in sqlite::list_databases(home)? {
                println!("{name}");
            }
            Ok(())
        }
        SqlCommands::Create { name } => {
            let path = sqlite::create_db(home, &name)?;
            println!("{}", path.display());
            Ok(())
        }
        SqlCommands::Tables { database, schema } => {
            for table in sqlite::list_tables(home, &database)? {
                println!("{}", table.name);
                if schema {
                    for col in &table.columns {
                        if col.decl_type.is_empty() {
                            println!("  {}", col.name);
                        } else {
                            println!("  {} {}", col.name, col.decl_type);
                        }
                    }
                }
            }
            Ok(())
        }
        SqlCommands::Exec { database, sql } => match sqlite::exec_sql(home, &database, &sql)? {
            SqlOutcome::Query(result) => {
                if !result.columns.is_empty() {
                    println!("{}", result.columns.join("\t"));
                }
                for row in result.rows {
                    println!("{}", row.join("\t"));
                }
                Ok(())
            }
            SqlOutcome::Exec(result) => {
                println!("ok {}", result.rows_affected);
                Ok(())
            }
        },
    }
}

fn dispatch_triple(home: &UnifierHome, cmd: TripleCommands) -> Result<()> {
    match cmd {
        TripleCommands::Add {
            subject,
            predicate,
            object,
        } => {
            sqlite::insert_triple(home, &subject, &predicate, &object)?;
            Ok(())
        }
        TripleCommands::Query {
            subject,
            predicate,
            object,
        } => {
            let rows = sqlite::query_triples(
                home,
                subject.as_deref(),
                predicate.as_deref(),
                object.as_deref(),
            )?;
            for (s, p, o) in rows {
                println!("{s}\t{p}\t{o}");
            }
            Ok(())
        }
    }
}

fn dispatch_data(home: &UnifierHome, cmd: Commands, ns: Option<&str>) -> Result<()> {
    match cmd {
        Commands::Put { key, value } => {
            let key = qualify(home, ns, key)?;
            put_key(home, &key, &value)?;
            Ok(())
        }
        Commands::Get { key } => {
            let key = qualify(home, ns, key)?;
            match get_key(home, &key)? {
                Some(v) => {
                    println!("{v}");
                    Ok(())
                }
                None => Err(crate::Error::msg(format!("key not found: {key}"))),
            }
        }
        Commands::Del { key } => {
            let key = qualify(home, ns, key)?;
            if delete_key(home, &key)? {
                Ok(())
            } else {
                Err(crate::Error::msg(format!("key not found: {key}")))
            }
        }
        Commands::Send {
            from,
            recipient,
            message,
        } => {
            let from = from.unwrap_or_else(|| crate::envelope::DEFAULT_SENDER.to_string());
            let id = send_from(home, &from, &recipient, &message)?;
            println!("{}", id.hyphenated());
            Ok(())
        }
        Commands::Cron { schedule, message } => {
            let id = post_cron(home, &schedule, &message)?;
            println!("{}", id.hyphenated());
            Ok(())
        }
        Commands::Poll {
            recipient,
            ack: do_ack,
        } => {
            let messages = poll_mailbox(home, &recipient)?;
            print_messages(&messages);
            if do_ack {
                for msg in &messages {
                    ack(home, &msg.id.hyphenated().to_string())?;
                }
            }
            Ok(())
        }
        Commands::PollCron { ack: do_ack } => {
            let messages = poll_cron(home)?;
            print_messages(&messages);
            if do_ack {
                for msg in &messages {
                    ack(home, &msg.id.hyphenated().to_string())?;
                }
            }
            Ok(())
        }
        Commands::List { path } => {
            let messages = list_dir(home, &path)?;
            print_messages(&messages);
            Ok(())
        }
        Commands::Ack { id_or_path } => {
            if ack(home, &id_or_path)? {
                Ok(())
            } else {
                Err(crate::Error::msg(format!(
                    "message not found: {id_or_path}"
                )))
            }
        }
        Commands::Root => {
            println!("{}", home.path().display());
            Ok(())
        }
        Commands::Event { .. } | Commands::Message { .. } | Commands::Tick(_) => {
            Err(crate::Error::msg(
                "event, message, and tick commands require the hot daemon; omit --no-daemon",
            ))
        }
        Commands::Daemon(_)
        | Commands::Chroot(_)
        | Commands::Namespace(_)
        | Commands::Sql(_)
        | Commands::Triple(_)
        | Commands::Log(_)
        | Commands::Serve { .. }
        | Commands::Web(_) => {
            unreachable!("handled in run()")
        }
    }
}

#[cfg(unix)]
fn dispatch_via_daemon(home: &UnifierHome, cmd: Commands, ns: Option<&str>) -> Result<()> {
    let mut client = Client::connect(home)?;
    match cmd {
        Commands::Put { key, value } => {
            let key = qualify(home, ns, key)?;
            response_ok(client.request(Request::Put { key, value })?)?;
            Ok(())
        }
        Commands::Get { key } => {
            let key = qualify(home, ns, key)?;
            match response_value(client.request(Request::Get { key })?)? {
                Some(v) => {
                    println!("{v}");
                    Ok(())
                }
                None => Err(crate::Error::msg("key not found")),
            }
        }
        Commands::Del { key } => {
            let key = qualify(home, ns, key)?;
            response_ok(client.request(Request::Del { key })?)?;
            Ok(())
        }
        Commands::Send {
            from,
            recipient,
            message,
        } => {
            let id = response_uuid(client.request(Request::Send {
                from,
                recipient,
                message,
            })?)?;
            println!("{}", id.hyphenated());
            Ok(())
        }
        Commands::Cron { schedule, message } => {
            let id = response_uuid(client.request(Request::Cron { schedule, message })?)?;
            println!("{}", id.hyphenated());
            Ok(())
        }
        Commands::Poll {
            recipient,
            ack: do_ack,
        } => {
            let messages = response_messages(client.request(Request::Poll { recipient })?)?;
            print_messages(&messages);
            if do_ack {
                for msg in &messages {
                    let found = response_found(client.request(Request::Ack {
                        id_or_path: msg.id.hyphenated().to_string(),
                    })?)?;
                    if !found {
                        return Err(crate::Error::msg(format!(
                            "message not found: {}",
                            msg.id.hyphenated()
                        )));
                    }
                }
            }
            Ok(())
        }
        Commands::PollCron { ack: do_ack } => {
            let messages = response_messages(client.request(Request::PollCron)?)?;
            print_messages(&messages);
            if do_ack {
                for msg in &messages {
                    response_found(client.request(Request::Ack {
                        id_or_path: msg.id.hyphenated().to_string(),
                    })?)?;
                }
            }
            Ok(())
        }
        Commands::List { path } => {
            let messages = response_messages(client.request(Request::List { path })?)?;
            print_messages(&messages);
            Ok(())
        }
        Commands::Ack { id_or_path } => {
            if response_found(client.request(Request::Ack { id_or_path })?)? {
                Ok(())
            } else {
                Err(crate::Error::msg("message not found"))
            }
        }
        Commands::Root => {
            println!("{}", home.path().display());
            Ok(())
        }
        Commands::Event { payload, ttl } => {
            let id = response_uuid(client.request(Request::Event { payload, ttl })?)?;
            println!("{}", id.hyphenated());
            Ok(())
        }
        Commands::Message {
            from,
            recipient,
            payload,
        } => {
            let id = response_uuid(client.request(Request::AgentMessage {
                from,
                to: recipient,
                payload,
            })?)?;
            println!("{}", id.hyphenated());
            Ok(())
        }
        Commands::Tick(cmd) => dispatch_tick_via_daemon(home, &mut client, cmd, ns),
        Commands::Daemon(_)
        | Commands::Chroot(_)
        | Commands::Namespace(_)
        | Commands::Sql(_)
        | Commands::Triple(_)
        | Commands::Log(_)
        | Commands::Serve { .. }
        | Commands::Web(_) => {
            unreachable!("handled in run()")
        }
    }
}

#[cfg(unix)]
fn dispatch_tick_via_daemon(
    home: &UnifierHome,
    client: &mut Client,
    cmd: TickCommands,
    ns: Option<&str>,
) -> Result<()> {
    use crate::daemon::{response_found, response_ok, Request, Response};

    // Prefer the dedicated tick control socket; fall back to the main daemon sock.
    let mut tick_client = Client::connect_tick(home).ok();
    let client = tick_client.as_mut().unwrap_or(client);

    match cmd {
        TickCommands::Start { label } => {
            let resp = client.request(Request::TickStart { label })?;
            match resp {
                Response::Ok {
                    tick: Some(t),
                    phase,
                    label,
                    ..
                } => {
                    let phase = phase.unwrap_or_else(|| "start".into());
                    if let Some(label) = label {
                        println!("tick {t} started phase={phase} label={label}");
                    } else {
                        println!("tick {t} started phase={phase}");
                    }
                    Ok(())
                }
                Response::Ok {
                    queued: Some(pos), ..
                } => {
                    println!("tick start queued at position {pos}");
                    Ok(())
                }
                Response::Err { error } => Err(crate::Error::msg(error)),
                _ => Err(crate::Error::msg("unexpected tick start response")),
            }
        }
        TickCommands::End => {
            let resp = client.request(Request::TickEnd)?;
            match resp {
                Response::Ok { tick: Some(t), .. } => {
                    println!("tick {t} committed");
                    Ok(())
                }
                Response::Err { error } => Err(crate::Error::msg(error)),
                _ => Err(crate::Error::msg("unexpected tick end response")),
            }
        }
        TickCommands::Status => {
            let resp = client.request(Request::TickStatus)?;
            match resp {
                Response::Ok { value: Some(v), .. } => {
                    println!("{v}");
                    Ok(())
                }
                Response::Err { error } => Err(crate::Error::msg(error)),
                _ => Err(crate::Error::msg("unexpected tick status response")),
            }
        }
        TickCommands::Phase { phase } => {
            let resp = client.request(Request::TickPhase { phase })?;
            match resp {
                Response::Ok {
                    tick: Some(t),
                    phase: Some(p),
                    ..
                } => {
                    println!("tick {t} phase={p}");
                    Ok(())
                }
                Response::Err { error } => Err(crate::Error::msg(error)),
                _ => Err(crate::Error::msg("unexpected tick phase response")),
            }
        }
        TickCommands::Lock { key } => {
            let key = qualify(home, ns, key)?;
            response_ok(client.request(Request::TickLock { key })?)?;
            Ok(())
        }
        TickCommands::Unlock { key } => {
            let key = qualify(home, ns, key)?;
            if response_found(client.request(Request::TickUnlock { key })?)? {
                Ok(())
            } else {
                Err(crate::Error::msg("lock not held"))
            }
        }
    }
}

struct ServeOpts {
    name: Option<String>,
    file: Option<std::path::PathBuf>,
    content_type: Option<String>,
    ttl: Option<u64>,
    wrap: bool,
    title: String,
    no_daemon: bool,
}

fn dispatch_serve(home: &UnifierHome, opts: ServeOpts) -> Result<()> {
    #[cfg(not(unix))]
    {
        let _ = (home, opts);
        return Err(crate::Error::msg("web serve requires a Unix platform"));
    }

    #[cfg(unix)]
    {
        use std::io::Read;
        use uuid::Uuid;

        if !opts.no_daemon {
            daemon::ensure_running(home)?;
            wait_for_http_port(home)?;
        } else if daemon::www::base_url(home).is_none() {
            return Err(crate::Error::msg(
                "web server is not listening; omit --no-daemon so the daemon can start it",
            ));
        }

        let mut raw = Vec::new();
        if let Some(path) = opts.file {
            raw = std::fs::read(&path)?;
        } else {
            std::io::stdin().read_to_end(&mut raw)?;
        }

        let body = if opts.wrap {
            let text = String::from_utf8_lossy(&raw);
            daemon::www::wrap_html(&opts.title, &text).into_bytes()
        } else {
            raw
        };

        let name = opts
            .name
            .unwrap_or_else(|| Uuid::new_v4().hyphenated().to_string());
        daemon::www::publish(home, &name, &body, opts.content_type.as_deref(), opts.ttl)?;
        let url = daemon::www::entry_url(home, &name)?;
        println!("{url}");
        Ok(())
    }
}

fn dispatch_web(home: &UnifierHome, cmd: WebCommands, no_daemon: bool) -> Result<()> {
    #[cfg(not(unix))]
    {
        let _ = (home, cmd, no_daemon);
        return Err(crate::Error::msg("web commands require a Unix platform"));
    }

    #[cfg(unix)]
    {
        if !no_daemon {
            daemon::ensure_running(home)?;
            wait_for_http_port(home)?;
        }

        match cmd {
            WebCommands::List => {
                if !no_daemon {
                    let mut client = Client::connect(home)?;
                    match client.request(Request::WebList)? {
                        crate::daemon::Response::Ok { value: Some(v), .. } => {
                            if !v.is_empty() {
                                println!("{v}");
                            }
                            Ok(())
                        }
                        crate::daemon::Response::Ok { .. } => Ok(()),
                        crate::daemon::Response::Err { error } => Err(crate::Error::msg(error)),
                    }
                } else {
                    for e in daemon::www::list(home)? {
                        let url = daemon::www::entry_url(home, &e.name).unwrap_or_default();
                        println!("{}\t{}\t{}\t{}", e.name, e.content_type, e.bytes, url);
                    }
                    Ok(())
                }
            }
            WebCommands::Url { name } => {
                let url = daemon::www::entry_url(home, &name)?;
                if daemon::www::load_meta(home, &name)?.is_none() {
                    return Err(crate::Error::msg(format!("web file not found: {name}")));
                }
                println!("{url}");
                Ok(())
            }
            WebCommands::KeyUrl { key } => {
                let url = daemon::www::key_url(home, &key)?;
                println!("{url}");
                Ok(())
            }
            WebCommands::Rm { name } => {
                let found = if !no_daemon {
                    let mut client = Client::connect(home)?;
                    response_found(client.request(Request::WebRm { name: name.clone() })?)?
                } else {
                    daemon::www::remove(home, &name)?
                };
                if found {
                    Ok(())
                } else {
                    Err(crate::Error::msg(format!("web file not found: {name}")))
                }
            }
            WebCommands::Status => {
                let url = if !no_daemon {
                    let mut client = Client::connect(home)?;
                    response_value(client.request(Request::WebStatus)?)?
                        .ok_or_else(|| crate::Error::msg("web server is not listening"))?
                } else {
                    daemon::www::base_url(home)
                        .ok_or_else(|| crate::Error::msg("web server is not listening"))?
                };
                println!("{url}");
                Ok(())
            }
        }
    }
}

#[cfg(unix)]
fn wait_for_http_port(home: &UnifierHome) -> Result<()> {
    for _ in 0..100 {
        if daemon::www::base_url(home).is_some() {
            return Ok(());
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
    Err(crate::Error::msg("web server failed to start"))
}

fn dispatch_log(home: &UnifierHome, cmd: LogCommands) -> Result<()> {
    match cmd {
        LogCommands::Start {
            name,
            parent,
            fields,
        } => {
            let parent_id = parent.as_deref().map(parse_uuid).transpose()?;
            let fields = spanlog::parse_fields(&fields)?;
            let id = spanlog::span_start(home, &name, parent_id, fields)?;
            println!("{}", id.hyphenated());
            Ok(())
        }
        LogCommands::End { id } => {
            let id = parse_uuid(&id)?;
            spanlog::span_end(home, id)?;
            println!("span {id} ended");
            Ok(())
        }
        LogCommands::Event {
            span,
            message,
            fields,
        } => {
            let span_id = parse_uuid(&span)?;
            let fields = spanlog::parse_fields(&fields)?;
            spanlog::log_event(home, span_id, &message, fields)?;
            Ok(())
        }
        LogCommands::Field { span, fields } => {
            let span_id = parse_uuid(&span)?;
            let fields = spanlog::parse_fields(&fields)?;
            spanlog::span_set_fields(home, span_id, fields)?;
            Ok(())
        }
        LogCommands::Tree { span } => {
            let root_id = span.as_deref().map(parse_uuid).transpose()?;
            let spans = spanlog::load_all_spans(home)?;
            let tree = spanlog::render_tree(&spans, root_id);
            print!("{tree}");
            Ok(())
        }
        LogCommands::List => {
            let spans = spanlog::load_all_spans(home)?;
            for span in &spans {
                let status = if span.ended_at.is_some() {
                    "ended"
                } else {
                    "open"
                };
                println!("{}\t{}\t{}", span.id.hyphenated(), span.name, status);
            }
            Ok(())
        }
    }
}

fn parse_uuid(s: &str) -> Result<uuid::Uuid> {
    uuid::Uuid::parse_str(s).map_err(|_| crate::Error::msg(format!("invalid UUID: {s}")))
}

fn print_messages(messages: &[Message]) {
    for (i, msg) in messages.iter().enumerate() {
        println!("{} {}", msg.id.hyphenated(), msg.path.display());
        println!("{}", msg.body);
        if i + 1 < messages.len() {
            println!("---");
        }
    }
}