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};
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!("---");
}
}
}