pub mod export;
use core_api::schema::Schema;
use core_api::{
default_max_edges, is_write_query, wal_commit_count_at, AlgoDir, BackupReport, DegreeConfig,
Explanation, GraphDb, IngestOptions, PageRankConfig, Predicate, ResultSet, RuleDef,
RuleSuggestion, SharedDb, SnapshotOptions, Stats, Value, WccConfig,
};
use export::ExportFormat;
use std::collections::BTreeMap;
use std::fmt::Write as _;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;
pub const N_ORGS: usize = 10;
pub const N_PROJECTS: usize = 20;
pub const N_PEOPLE: usize = 30;
pub const SAMPLE_QUERY: &str = "\
MATCH (p:Person {id: 'person-01'})-[r:FIT]->(proj:Project)
RETURN p, proj, r.score AS score
ORDER BY score DESC, proj";
const SAMPLE_EXPLAIN_A: &str = "person-01";
const SAMPLE_EXPLAIN_B: &str = "proj-01";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ServeUi {
Filesystem(PathBuf),
Embedded,
None,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AlgoSubcmd {
Pagerank,
Wcc,
Degree,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Command {
Serve {
db_dir: PathBuf,
addr: SocketAddr,
ui: ServeUi,
demo_if_empty: bool,
token: Option<String>,
role_tokens: Vec<(String, String)>,
snapshot_every: Option<Duration>,
},
Mcp {
db_dir: PathBuf,
},
Stats {
db_dir: PathBuf,
},
Demo {
db_dir: PathBuf,
},
AsOf {
db_dir: PathBuf,
commit: u64,
query: Option<String>,
},
Suggest {
db_dir: PathBuf,
},
Algo {
db_dir: PathBuf,
subcmd: AlgoSubcmd,
top: usize,
},
Query {
db_dir: PathBuf,
cypher: String,
},
Snapshot {
db_dir: PathBuf,
keep_wal: bool,
archive_wal: bool,
retention: Option<u32>,
},
SchemaApply {
db_dir: PathBuf,
schema_file: PathBuf,
},
Migrate {
db_dir: PathBuf,
},
Verify {
db_dir: PathBuf,
},
Backup {
db_dir: PathBuf,
dest: PathBuf,
},
Export {
db_dir: PathBuf,
dest: PathBuf,
format: ExportFormat,
},
Help,
}
#[derive(Debug)]
pub struct DemoOutcome {
pub auto_fk_rules: Vec<String>,
pub sample_query: String,
pub sample_result: ResultSet,
pub explanations: Vec<Explanation>,
pub stats: Stats,
pub suggestion: Option<RuleSuggestion>,
}
#[derive(Debug)]
pub struct CliError(pub String);
impl std::fmt::Display for CliError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl std::error::Error for CliError {}
impl From<core_api::GraphError> for CliError {
fn from(e: core_api::GraphError) -> Self {
CliError(e.to_string())
}
}
impl From<std::io::Error> for CliError {
fn from(e: std::io::Error) -> Self {
CliError(e.to_string())
}
}
pub fn usage() -> &'static str {
"\
mushroomdb — embedded graph database
Usage:
mushroomdb serve <db-dir> [--addr 127.0.0.1:8080] [--token <secret>] [--ui <dist-dir>] [--no-ui] [--demo-if-empty] [--snapshot-every <secs>]
mushroomdb mcp <db-dir>
mushroomdb stats <db-dir>
mushroomdb demo <db-dir>
mushroomdb suggest <db-dir>
mushroomdb asof <db-dir> --commit N [--query \"MATCH ...\"]
mushroomdb query <db-dir> [--query \"MATCH ...\"] <cypher…>
mushroomdb snapshot <db-dir> [--keep-wal]
mushroomdb migrate <db-dir>
mushroomdb verify <db-dir> validate CRC32 integrity of every snapshot section
mushroomdb backup <db-dir> <dest> process-local consistent copy of the database to <dest>
WARNING: unsafe against a concurrently running serve process;
use POST /backup on the HTTP server for live-serve backups
mushroomdb export <db-dir> <dest> --format jsonl|parquet export all data
mushroomdb schema apply <db-dir> <schema.json>
mushroomdb algo pagerank <db-dir> [--top N]
mushroomdb algo wcc <db-dir> [--top N]
mushroomdb algo degree <db-dir> [--top N]
mushroomdb --help
Default serve address is 127.0.0.1:8080. Non-loopback --addr requires --token or MUSHROOMDB_TOKEN.
"
}
pub fn parse_args<S: AsRef<str>>(args: &[S]) -> Result<Command, String> {
let args: Vec<&str> = args.iter().map(AsRef::as_ref).collect();
if args.is_empty() {
return Ok(Command::Help);
}
match args[0] {
"--help" | "-h" | "help" => Ok(Command::Help),
"serve" => parse_serve(&args[1..]),
"mcp" => parse_one_dir("mcp", &args[1..]).map(|db_dir| Command::Mcp { db_dir }),
"stats" => parse_one_dir("stats", &args[1..]).map(|db_dir| Command::Stats { db_dir }),
"demo" => parse_one_dir("demo", &args[1..]).map(|db_dir| Command::Demo { db_dir }),
"suggest" => parse_one_dir("suggest", &args[1..]).map(|db_dir| Command::Suggest { db_dir }),
"asof" => parse_asof(&args[1..]),
"algo" => parse_algo(&args[1..]),
"query" => parse_query(&args[1..]),
"snapshot" => parse_snapshot(&args[1..]),
"schema" => parse_schema(&args[1..]),
"migrate" => parse_one_dir("migrate", &args[1..]).map(|db_dir| Command::Migrate { db_dir }),
"verify" => parse_one_dir("verify", &args[1..]).map(|db_dir| Command::Verify { db_dir }),
"backup" => parse_backup(&args[1..]),
"export" => parse_export(&args[1..]),
other => Err(format!("unknown command: {other}")),
}
}
fn default_addr() -> SocketAddr {
SocketAddr::from(([127, 0, 0, 1], 8080))
}
fn parse_serve(args: &[&str]) -> Result<Command, String> {
let mut db_dir = None;
let mut addr = default_addr();
let mut ui = ServeUi::Embedded;
let mut saw_ui = false;
let mut saw_no_ui = false;
let mut demo_if_empty = false;
let mut token = None;
let mut role_tokens: Vec<(String, String)> = Vec::new();
let mut snapshot_every = None;
let mut i = 0;
while i < args.len() {
let a = args[i];
if a == "--addr" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --addr".to_string())?;
addr = val.parse().map_err(|_| format!("invalid address: {val}"))?;
i += 2;
} else if let Some(val) = a.strip_prefix("--addr=") {
addr = val.parse().map_err(|_| format!("invalid address: {val}"))?;
i += 1;
} else if a == "--ui" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --ui".to_string())?;
ui = ServeUi::Filesystem(PathBuf::from(val));
saw_ui = true;
i += 2;
} else if let Some(val) = a.strip_prefix("--ui=") {
ui = ServeUi::Filesystem(PathBuf::from(val));
saw_ui = true;
i += 1;
} else if a == "--no-ui" {
ui = ServeUi::None;
saw_no_ui = true;
i += 1;
} else if a == "--demo-if-empty" {
demo_if_empty = true;
i += 1;
} else if a == "--token" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --token".to_string())?;
token = Some(val.to_string());
i += 2;
} else if let Some(val) = a.strip_prefix("--token=") {
token = Some(val.to_string());
i += 1;
} else if a == "--role-token" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --role-token".to_string())?;
let (tok, role) = parse_role_token(val)?;
role_tokens.push((tok, role));
i += 2;
} else if let Some(val) = a.strip_prefix("--role-token=") {
let (tok, role) = parse_role_token(val)?;
role_tokens.push((tok, role));
i += 1;
} else if a == "--snapshot-every" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --snapshot-every".to_string())?;
snapshot_every = Some(parse_snapshot_every(val)?);
i += 2;
} else if let Some(val) = a.strip_prefix("--snapshot-every=") {
snapshot_every = Some(parse_snapshot_every(val)?);
i += 1;
} else if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
} else if db_dir.is_none() {
db_dir = Some(PathBuf::from(a));
i += 1;
} else {
return Err(format!("unexpected extra argument: {a}"));
}
}
if saw_ui && saw_no_ui {
return Err("cannot combine --ui and --no-ui".to_string());
}
let db_dir = db_dir.ok_or_else(|| "serve requires <db-dir>".to_string())?;
Ok(Command::Serve {
db_dir,
addr,
ui,
demo_if_empty,
token,
role_tokens,
snapshot_every,
})
}
fn parse_role_token(val: &str) -> Result<(String, String), String> {
let (tok, role) = val
.split_once(':')
.ok_or_else(|| format!("--role-token requires TOKEN:ROLE format, got: {val}"))?;
if tok.is_empty() {
return Err("--role-token: TOKEN must not be empty".to_string());
}
if role.is_empty() {
return Err("--role-token: ROLE must not be empty".to_string());
}
Ok((tok.to_string(), role.to_string()))
}
fn parse_snapshot_every(val: &str) -> Result<Duration, String> {
let secs: u64 = val
.parse()
.map_err(|_| format!("invalid --snapshot-every: {val}"))?;
if secs == 0 {
return Err("--snapshot-every must be a positive number of seconds".into());
}
Ok(Duration::from_secs(secs))
}
pub fn validate_ui_dir(dir: &Path) -> Result<PathBuf, String> {
if !dir.is_dir() {
return Err(format!("--ui directory does not exist: {}", dir.display()));
}
let index = dir.join("index.html");
if !index.is_file() {
return Err(format!(
"--ui directory is missing index.html: {}",
dir.display()
));
}
Ok(dir.to_path_buf())
}
fn parse_asof(args: &[&str]) -> Result<Command, String> {
let mut db_dir = None;
let mut commit: Option<u64> = None;
let mut query: Option<String> = None;
let mut i = 0;
while i < args.len() {
let a = args[i];
if a == "--commit" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --commit".to_string())?;
commit = Some(
val.parse()
.map_err(|_| format!("invalid commit index: {val}"))?,
);
i += 2;
} else if let Some(val) = a.strip_prefix("--commit=") {
commit = Some(
val.parse()
.map_err(|_| format!("invalid commit index: {val}"))?,
);
i += 1;
} else if a == "--query" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --query".to_string())?;
query = Some(val.to_string());
i += 2;
} else if let Some(val) = a.strip_prefix("--query=") {
query = Some(val.to_string());
i += 1;
} else if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
} else if db_dir.is_none() {
db_dir = Some(PathBuf::from(a));
i += 1;
} else {
return Err(format!("unexpected extra argument: {a}"));
}
}
let db_dir = db_dir.ok_or_else(|| "asof requires <db-dir>".to_string())?;
let commit = commit.ok_or_else(|| "asof requires --commit N".to_string())?;
Ok(Command::AsOf {
db_dir,
commit,
query,
})
}
pub fn run_asof(db_dir: &Path, commit: u64, query: Option<&str>) -> Result<String, CliError> {
let total = wal_commit_count_at(db_dir)?;
let db = GraphDb::open_at(db_dir, commit)?;
let mut out = String::new();
let _ = writeln!(out, "as-of commit {} of {}", commit, total);
if let Some(cypher) = query {
let params = BTreeMap::new();
let rs = db.query(cypher, ¶ms)?;
out.push_str(&format_result_set(&rs));
}
Ok(out)
}
fn parse_query(args: &[&str]) -> Result<Command, String> {
let mut db_dir = None;
let mut query_flag: Option<String> = None;
let mut cypher_parts: Vec<&str> = Vec::new();
let mut i = 0;
while i < args.len() {
let a = args[i];
if a == "--query" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --query".to_string())?;
query_flag = Some(val.to_string());
i += 2;
} else if let Some(val) = a.strip_prefix("--query=") {
query_flag = Some(val.to_string());
i += 1;
} else if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
} else if db_dir.is_none() {
db_dir = Some(PathBuf::from(a));
i += 1;
} else {
cypher_parts.push(a);
i += 1;
}
}
let db_dir = db_dir.ok_or_else(|| "query requires <db-dir>".to_string())?;
let cypher = if let Some(q) = query_flag {
if !cypher_parts.is_empty() {
return Err(
"query: pass Cypher as remaining arguments or --query, not both".to_string(),
);
}
q
} else {
if cypher_parts.is_empty() {
return Err("query requires a Cypher string".to_string());
}
cypher_parts.join(" ")
};
Ok(Command::Query { db_dir, cypher })
}
pub fn run_query(db_dir: &Path, cypher: &str) -> Result<String, CliError> {
let params = BTreeMap::new();
let is_write = is_write_query(cypher).map_err(CliError)?;
let rs = if is_write {
let mut db = GraphDb::open(db_dir)?;
db.query_write(cypher, ¶ms)?
} else {
let db = GraphDb::open(db_dir)?;
db.query(cypher, ¶ms)?
};
Ok(format_result_set(&rs))
}
fn parse_snapshot(args: &[&str]) -> Result<Command, String> {
let mut db_dir = None;
let mut keep_wal = false;
let mut archive_wal = false;
let mut retention: Option<u32> = None;
let mut i = 0;
while i < args.len() {
let a = args[i];
if a == "--keep-wal" {
keep_wal = true;
i += 1;
} else if a == "--archive-wal" {
archive_wal = true;
i += 1;
} else if a.starts_with("--retention=") {
let v = a.trim_start_matches("--retention=");
retention = Some(
v.parse::<u32>()
.map_err(|_| format!("--retention= expects a u32, got: {v}"))?,
);
i += 1;
} else if a == "--retention" {
i += 1;
let v = args
.get(i)
.ok_or_else(|| "--retention requires a value".to_string())?;
retention = Some(
v.parse::<u32>()
.map_err(|e| format!("--retention value error: {e}"))?,
);
i += 1;
} else if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
} else if db_dir.is_none() {
db_dir = Some(PathBuf::from(a));
i += 1;
} else {
return Err(format!("unexpected extra argument: {a}"));
}
}
let db_dir = db_dir.ok_or_else(|| "snapshot requires <db-dir>".to_string())?;
Ok(Command::Snapshot {
db_dir,
keep_wal,
archive_wal,
retention,
})
}
pub fn run_migrate(db_dir: &Path) -> Result<String, CliError> {
let current = core_api::SNAPSHOT_VERSION;
let from_ver = core_api::snapshot_version_at(db_dir)?;
if from_ver == Some(current) {
return Ok(format!("already current (V{current})\n"));
}
if from_ver.is_some() {
std::fs::copy(db_dir.join("snapshot.bin"), db_dir.join("snapshot.bin.bak"))?;
}
let mut db = GraphDb::open_with_options(
db_dir,
core_api::OpenOptions {
auto_migrate: false,
},
)?;
db.snapshot()?;
let msg = match from_ver {
Some(ver) => format!("migrated V{ver} -> V{current}\n"),
None => format!("migrated WAL-only -> V{current}\n"),
};
Ok(msg)
}
pub fn run_verify(db_dir: &Path) -> Result<String, CliError> {
let results = core_api::verify_snapshot(db_dir)
.map_err(|e| CliError(format!("verify: cannot open snapshot: {e}")))?;
let mut any_fail = false;
let mut out = String::new();
for (id, section_name, byte_len, result) in &results {
match result {
Ok(()) => {
let _ = writeln!(
out,
" section {:2} ({:<12}) {:>10} bytes OK",
id, section_name, byte_len
);
}
Err(msg) => {
let _ = writeln!(
out,
" section {:2} ({:<12}) {:>10} bytes CORRUPT: {msg}",
id, section_name, byte_len
);
any_fail = true;
}
}
}
if any_fail {
Err(CliError(format!("integrity check FAILED:\n{out}")))
} else {
Ok(format!(
"integrity check OK ({} sections):\n{out}",
results.len()
))
}
}
pub fn run_snapshot(
db_dir: &Path,
keep_wal: bool,
archive_wal: bool,
retention: Option<u32>,
) -> Result<String, CliError> {
let mut db = GraphDb::open(db_dir)?;
if archive_wal {
db.set_wal_archive_retention(retention);
db.snapshot_with(SnapshotOptions {
archive_wal: true,
keep_wal: false,
})?;
} else if keep_wal {
db.snapshot_with(SnapshotOptions {
keep_wal: true,
archive_wal: false,
})?;
} else {
db.snapshot()?;
}
Ok(format!(
"snapshot written: {}\n",
db_dir.join("snapshot.bin").display()
))
}
fn parse_schema(args: &[&str]) -> Result<Command, String> {
if args.is_empty() {
return Err("schema requires a subcommand: apply".to_string());
}
match args[0] {
"apply" => parse_schema_apply(&args[1..]),
other => Err(format!(
"unknown schema subcommand: {other}; expected apply"
)),
}
}
fn parse_schema_apply(args: &[&str]) -> Result<Command, String> {
let mut db_dir = None;
let mut schema_file = None;
for a in args {
if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
}
if db_dir.is_none() {
db_dir = Some(PathBuf::from(*a));
} else if schema_file.is_none() {
schema_file = Some(PathBuf::from(*a));
} else {
return Err(format!("unexpected extra argument: {a}"));
}
}
let db_dir = db_dir.ok_or_else(|| "schema apply requires <db-dir>".to_string())?;
let schema_file =
schema_file.ok_or_else(|| "schema apply requires <schema.json>".to_string())?;
Ok(Command::SchemaApply {
db_dir,
schema_file,
})
}
pub fn run_schema_apply(db_dir: &Path, schema_file: &Path) -> Result<String, CliError> {
let json = std::fs::read_to_string(schema_file)
.map_err(|e| CliError(format!("cannot read {}: {e}", schema_file.display())))?;
let schema: Schema = serde_json::from_str(&json).map_err(|e| {
CliError(format!(
"invalid schema JSON in {}: {e}",
schema_file.display()
))
})?;
let mut db = GraphDb::open(db_dir)?;
let diff = db.apply_schema(&schema)?;
let mut out = String::new();
for entry in &diff.created {
let _ = writeln!(out, "created {entry}");
}
for entry in &diff.updated {
let _ = writeln!(out, "updated {entry}");
}
for entry in &diff.unchanged {
let _ = writeln!(out, "unchanged {entry}");
}
if diff.created.is_empty() && diff.updated.is_empty() && diff.unchanged.is_empty() {
let _ = writeln!(out, "schema applied: nothing to do (empty schema)");
}
Ok(out)
}
fn parse_backup(args: &[&str]) -> Result<Command, String> {
let mut db_dir = None;
let mut dest = None;
for a in args {
if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
}
if db_dir.is_none() {
db_dir = Some(PathBuf::from(*a));
} else if dest.is_none() {
dest = Some(PathBuf::from(*a));
} else {
return Err(format!("unexpected extra argument: {a}"));
}
}
let db_dir = db_dir.ok_or_else(|| "backup requires <db-dir>".to_string())?;
let dest = dest.ok_or_else(|| "backup requires <dest>".to_string())?;
Ok(Command::Backup { db_dir, dest })
}
fn parse_export(args: &[&str]) -> Result<Command, String> {
let mut db_dir = None;
let mut dest = None;
let mut format = ExportFormat::Jsonl;
let mut i = 0;
while i < args.len() {
let a = args[i];
if a == "--format" {
let val = args
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --format".to_string())?;
format = ExportFormat::parse(val)
.ok_or_else(|| format!("unknown format '{val}'; expected jsonl or parquet"))?;
i += 2;
} else if let Some(val) = a.strip_prefix("--format=") {
format = ExportFormat::parse(val)
.ok_or_else(|| format!("unknown format '{val}'; expected jsonl or parquet"))?;
i += 1;
} else if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
} else if db_dir.is_none() {
db_dir = Some(PathBuf::from(a));
i += 1;
} else if dest.is_none() {
dest = Some(PathBuf::from(a));
i += 1;
} else {
return Err(format!("unexpected extra argument: {a}"));
}
}
let db_dir = db_dir.ok_or_else(|| "export requires <db-dir>".to_string())?;
let dest = dest.ok_or_else(|| "export requires <dest>".to_string())?;
Ok(Command::Export {
db_dir,
dest,
format,
})
}
pub fn run_backup(db_dir: &Path, dest: &Path) -> Result<BackupReport, CliError> {
let db = GraphDb::open(db_dir)?;
Ok(db.backup_to(dest)?)
}
pub fn format_backup(dest: &Path, report: &BackupReport) -> String {
let mut out = String::new();
writeln!(out, "backup to: {}", dest.display()).unwrap();
writeln!(out, " files: {}", report.files.join(", ")).unwrap();
writeln!(out, " bytes: {}", report.bytes).unwrap();
writeln!(out, " verified: {}", report.verified).unwrap();
out
}
pub fn run_export(db_dir: &Path, dest: &Path, format: &ExportFormat) -> Result<String, CliError> {
let db = GraphDb::open(db_dir)?;
let nodes = db.all_nodes_for_export();
let edges = db.all_edges_for_export();
let mut rules = db.rules();
rules.sort_by(|a, b| a.name.cmp(&b.name));
let node_count = nodes.len();
let edge_count = edges.len();
let rule_count = rules.len();
match format {
ExportFormat::Jsonl => export::write_jsonl(&nodes, &edges, &rules, dest)?,
ExportFormat::Parquet => export::write_parquet(&nodes, &edges, &rules, dest)?,
}
Ok(format!(
"exported to {} (format={}): {} nodes, {} edges, {} rules\n",
dest.display(),
format.name(),
node_count,
edge_count,
rule_count
))
}
fn format_result_set(rs: &ResultSet) -> String {
let mut out = String::new();
let _ = writeln!(out, "columns: {}", rs.columns().join(", "));
for i in 0..rs.len() {
let cells: Vec<String> = rs
.columns()
.iter()
.map(|c| format!("{c}={}", fmt_cell(rs.get(i, c))))
.collect();
let _ = writeln!(out, " {}", cells.join(" "));
}
out
}
fn parse_algo(args: &[&str]) -> Result<Command, String> {
if args.is_empty() {
return Err("algo requires a subcommand: pagerank | wcc | degree".to_string());
}
let subcmd = match args[0] {
"pagerank" => AlgoSubcmd::Pagerank,
"wcc" => AlgoSubcmd::Wcc,
"degree" => AlgoSubcmd::Degree,
other => {
return Err(format!(
"unknown algo subcommand: {other}; expected pagerank | wcc | degree"
))
}
};
let rest = &args[1..];
let mut db_dir = None;
let mut top: usize = 20;
let mut i = 0;
while i < rest.len() {
let a = rest[i];
if a == "--top" {
let val = rest
.get(i + 1)
.copied()
.ok_or_else(|| "missing value for --top".to_string())?;
top = val
.parse()
.map_err(|_| format!("--top must be a non-negative integer, got {val}"))?;
i += 2;
} else if let Some(val) = a.strip_prefix("--top=") {
top = val
.parse()
.map_err(|_| format!("--top must be a non-negative integer, got {val}"))?;
i += 1;
} else if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
} else if db_dir.is_none() {
db_dir = Some(PathBuf::from(a));
i += 1;
} else {
return Err(format!("unexpected extra argument: {a}"));
}
}
let db_dir = db_dir.ok_or_else(|| format!("algo {} requires <db-dir>", args[0]))?;
Ok(Command::Algo {
db_dir,
subcmd,
top,
})
}
pub fn run_algo(db_dir: &Path, subcmd: &AlgoSubcmd, top: usize) -> Result<String, CliError> {
let db = GraphDb::open(db_dir)?;
match subcmd {
AlgoSubcmd::Pagerank => {
let config = PageRankConfig::default();
let report = db.pagerank(&config);
Ok(format_pagerank(&report, top))
}
AlgoSubcmd::Wcc => {
let config = WccConfig::default();
let report = db.connected_components(&config);
Ok(format_wcc(&report, top))
}
AlgoSubcmd::Degree => {
let config = DegreeConfig {
direction: AlgoDir::Both,
..DegreeConfig::default()
};
let report = db.degree_centrality(&config);
Ok(format_degree(&report, top))
}
}
}
fn format_pagerank(report: &core_api::PageRankReport, top: usize) -> String {
let mut buf = String::new();
let _ = writeln!(buf, "== pagerank (converged={}) ==", report.converged);
let rows = if top == 0 {
report.scores.as_slice()
} else {
&report.scores[..top.min(report.scores.len())]
};
for (i, (key, score)) in rows.iter().enumerate() {
let _ = writeln!(buf, " {:>4} {:<40} {:.6}", i + 1, key, score);
}
buf
}
fn format_wcc(report: &core_api::WccReport, top: usize) -> String {
let mut buf = String::new();
let _ = writeln!(buf, "== wcc (truncated={}) ==", report.truncated);
let rows = if top == 0 {
report.components.as_slice()
} else {
&report.components[..top.min(report.components.len())]
};
for (key, comp_id) in rows {
let _ = writeln!(buf, " {:<40} component={}", key, comp_id);
}
buf
}
fn format_degree(report: &core_api::DegreeReport, top: usize) -> String {
let mut buf = String::new();
let _ = writeln!(
buf,
"== degree centrality (truncated={}) ==",
report.truncated
);
let rows = if top == 0 {
report.scores.as_slice()
} else {
&report.scores[..top.min(report.scores.len())]
};
for (i, (key, deg)) in rows.iter().enumerate() {
let _ = writeln!(buf, " {:>4} {:<40} degree={}", i + 1, key, deg);
}
buf
}
fn parse_one_dir(cmd: &str, args: &[&str]) -> Result<PathBuf, String> {
let mut db_dir = None;
for a in args {
if a.starts_with('-') {
return Err(format!("unexpected flag: {a}"));
}
if db_dir.is_some() {
return Err(format!("unexpected extra argument: {a}"));
}
db_dir = Some(PathBuf::from(*a));
}
db_dir.ok_or_else(|| format!("{cmd} requires <db-dir>"))
}
pub fn format_stats(stats: &Stats) -> String {
let mut out = String::new();
let _ = writeln!(
out,
"nodes: {} live, {} tombstoned",
stats.nodes_live, stats.nodes_tombstoned
);
let _ = writeln!(out, "edges: {}", stats.edges);
let _ = writeln!(out, "rules: {}", stats.rules.len());
for r in &stats.rules {
let _ = writeln!(
out,
" {:<28} edges={} tripped={}",
r.name, r.edges, r.tripped
);
}
out
}
pub fn read_stats(dir: &Path) -> Result<Stats, CliError> {
let db = SharedDb::open(dir)?;
let stats = db.read().stats();
Ok(stats)
}
pub fn run_demo(dir: &Path) -> Result<DemoOutcome, CliError> {
refuse_non_empty(dir)?;
let db = SharedDb::open(dir)?;
let opts = IngestOptions::default();
let mut auto_fk_rules = Vec::new();
{
let mut w = db.write();
for (label, json) in [
("Org", org_json()),
("Project", project_json()),
("Person", person_json()),
] {
let report = w.ingest_json(label, &json, &opts)?;
if !report.row_errors.is_empty() {
return Err(CliError(format!(
"demo ingest of {label} had row errors: {:?}",
report.row_errors
)));
}
auto_fk_rules.extend(report.rules_created);
}
let skill_fit = Predicate::Overlap {
field: "skills".into(),
min: 0.5,
};
let skill_fit_k = Some(default_max_edges(&skill_fit));
w.create_rule(RuleDef {
name: "skill_fit".into(),
src_label: "Person".into(),
dst_label: "Project".into(),
predicate: skill_fit,
edge_type: "FIT".into(),
weight_prop: Some("score".into()),
max_edges: skill_fit_k,
approximate: false,
via_label: None,
via_edge: None,
via_dir: None,
})?;
let founded_within = Predicate::NumericWithin {
field: "founded_year".into(),
tolerance: 2.0,
};
let founded_within_k = Some(default_max_edges(&founded_within));
w.create_rule(RuleDef {
name: "founded_within".into(),
src_label: "Org".into(),
dst_label: "Org".into(),
predicate: founded_within,
edge_type: "FOUNDED_WITHIN".into(),
weight_prop: Some("score".into()),
max_edges: founded_within_k,
approximate: false,
via_label: None,
via_edge: None,
via_dir: None,
})?;
let nearby_office = Predicate::GeoRadius {
field: "office".into(),
km: 50.0,
};
let nearby_office_k = Some(default_max_edges(&nearby_office));
w.create_rule(RuleDef {
name: "nearby_office".into(),
src_label: "Org".into(),
dst_label: "Org".into(),
predicate: nearby_office,
edge_type: "NEARBY_OFFICE".into(),
weight_prop: Some("score".into()),
max_edges: nearby_office_k,
approximate: false,
via_label: None,
via_edge: None,
via_dir: None,
})?;
let similar_interests = Predicate::VectorSimilar {
field: "embedding".into(),
min: 0.8,
};
let similar_interests_k = Some(default_max_edges(&similar_interests));
w.create_rule(RuleDef {
name: "similar_interests".into(),
src_label: "Person".into(),
dst_label: "Person".into(),
predicate: similar_interests,
edge_type: "SIMILAR".into(),
weight_prop: Some("score".into()),
max_edges: similar_interests_k,
approximate: false,
via_label: None,
via_edge: None,
via_dir: None,
})?;
}
let r = db.read();
let sample_result = r.query(SAMPLE_QUERY, &BTreeMap::new())?;
let explanations = r.explain(SAMPLE_EXPLAIN_A, SAMPLE_EXPLAIN_B)?;
let stats = r.stats();
let suggestion = r.suggest_rules().into_iter().next();
Ok(DemoOutcome {
auto_fk_rules,
sample_query: SAMPLE_QUERY.to_string(),
sample_result,
explanations,
stats,
suggestion,
})
}
fn dir_is_empty_or_absent(dir: &Path) -> Result<bool, CliError> {
if dir.is_file() {
return Err(CliError(format!(
"demo refuses a non-empty directory: {} is a file",
dir.display()
)));
}
if !dir.exists() {
return Ok(true);
}
Ok(std::fs::read_dir(dir)?.next().is_none())
}
fn refuse_non_empty(dir: &Path) -> Result<(), CliError> {
if dir_is_empty_or_absent(dir)? {
Ok(())
} else {
Err(CliError(format!(
"demo refuses a non-empty directory: {} \
(directory must be empty — including hidden files)",
dir.display()
)))
}
}
pub fn maybe_run_demo_if_empty(dir: &Path) -> Result<Option<DemoOutcome>, CliError> {
if dir_is_empty_or_absent(dir)? {
Ok(Some(run_demo(dir)?))
} else {
Ok(None)
}
}
fn json_array(rows: impl IntoIterator<Item = String>) -> String {
let mut out = String::from("[");
let mut first = true;
for row in rows {
if !first {
out.push(',');
}
first = false;
out.push_str(&row);
}
out.push(']');
out
}
fn wrap_proj(i: usize) -> usize {
(i - 1) % N_PROJECTS + 1
}
fn skill_window_json(start: usize, len: usize) -> String {
let parts: Vec<String> = (0..len)
.map(|k| format!(r#""s{:02}""#, wrap_proj(start + k)))
.collect();
format!("[{}]", parts.join(","))
}
fn org_office(i: usize) -> (f64, f64) {
match i {
1 => (40.7128, -74.0060), 2 => (48.8566, 2.3522), 3 => (51.5074, -0.1278), 4 => (37.7749, -122.4194), 5 => (37.8044, -122.2711), 6 => (37.8715, -122.2730), 7 => (40.7178, -74.0431), 8 => (51.4769, 0.0005), 9 => (48.8014, 2.1301), 10 => (40.7357, -74.1724), _ => unreachable!("demo orgs are 1..=10"),
}
}
fn person_embedding_json(i: usize) -> String {
let mut v = [0.0_f64; 8];
match i {
9 | 19 | 29 => {
v[0] = 0.8;
v[1] = 0.6;
}
10 | 20 | 30 => {
v[0] = 0.6;
v[1] = 0.8;
}
_ => {
let axis = (i - 1) % 10;
debug_assert!(axis < 8);
v[axis] = 1.0;
}
}
let parts: Vec<String> = v.iter().map(|x| format!("{x}")).collect();
format!("[{}]", parts.join(","))
}
fn org_json() -> String {
json_array((1..=N_ORGS).map(|i| {
let year = 2010 + (i as i64 - 1);
let (lat, lon) = org_office(i);
format!(
r#"{{"id":"org-{i:02}","name":"Org {i}","founded_year":{year},"office":[{lat},{lon}],"skills":{}}}"#,
skill_window_json(i, 3)
)
}))
}
fn project_json() -> String {
json_array((1..=N_PROJECTS).map(|i| {
let org = (i - 1) % N_ORGS + 1;
format!(
r#"{{"id":"proj-{i:02}","name":"Project {i}","org_id":"org-{org:02}","skills":{}}}"#,
skill_window_json(i, 3)
)
}))
}
fn person_json() -> String {
json_array((1..=N_PEOPLE).map(|i| {
let org = (i - 1) % N_ORGS + 1;
let proj = (i - 1) % N_PROJECTS + 1;
format!(
r#"{{"id":"person-{i:02}","name":"Person {i}","org_id":"org-{org:02}","project_id":"proj-{proj:02}","embedding":{},"skills":{}}}"#,
person_embedding_json(i),
skill_window_json(proj, 3)
)
}))
}
pub fn format_demo(dir: &Path, out: &DemoOutcome) -> String {
let mut buf = String::new();
let _ = writeln!(buf, "== demo ==");
let _ = writeln!(
buf,
"ingested {N_ORGS} Orgs, {N_PROJECTS} Projects, {N_PEOPLE} People"
);
let _ = writeln!(
buf,
"overlap rule: skill_fit (Person.skills ∩ Project.skills, min 0.5)"
);
let _ = writeln!(
buf,
"numeric rule: founded_within (Org.founded_year, tolerance 2)"
);
let _ = writeln!(buf, "geo rule: nearby_office (Org.office [lat,lon], 50 km)");
let _ = writeln!(
buf,
"vector rule: similar_interests (Person.embedding dim 8, min 0.8)"
);
let _ = writeln!(buf);
let _ = writeln!(buf, "== auto-FK rules ==");
let mut names = out.auto_fk_rules.clone();
names.sort();
for name in names {
let _ = writeln!(buf, " {name}");
}
let _ = writeln!(buf);
let _ = writeln!(buf, "== query ==");
let _ = writeln!(buf, "{}", out.sample_query);
let _ = writeln!(buf);
let _ = writeln!(buf, "columns: {}", out.sample_result.columns().join(", "));
for i in 0..out.sample_result.len() {
let cells: Vec<String> = out
.sample_result
.columns()
.iter()
.map(|c| format!("{c}={}", fmt_cell(out.sample_result.get(i, c))))
.collect();
let _ = writeln!(buf, " {}", cells.join(" "));
}
let _ = writeln!(buf);
let _ = writeln!(
buf,
"== explain ({SAMPLE_EXPLAIN_A}, {SAMPLE_EXPLAIN_B}) =="
);
for e in &out.explanations {
let weight = e
.weight
.map(|w| fmt_value(&Value::Float(w)))
.unwrap_or_else(|| "none".into());
let _ = writeln!(
buf,
" rule={} type={} {}→{} weight={}",
e.rule, e.edge_type, e.src_key, e.dst_key, weight
);
}
let _ = writeln!(buf);
let _ = writeln!(buf, "== serve ==");
let _ = writeln!(buf, " mushroomdb serve {}", dir.display());
if let Some(s) = &out.suggestion {
let _ = writeln!(buf);
let _ = writeln!(buf, "== suggested rule (teaser) ==");
let _ = writeln!(buf, " {}", s.def.name);
let _ = writeln!(
buf,
" {} → {} via {:?}",
s.def.src_label, s.def.dst_label, s.def.predicate
);
let _ = writeln!(buf, " est_edges: ~{}", s.est_edges);
let _ = writeln!(buf, " {}", s.rationale);
let _ = writeln!(
buf,
" (run `mushroomdb suggest {}` for full analysis)",
dir.display()
);
}
buf
}
pub fn run_suggest(dir: &Path) -> Result<Vec<RuleSuggestion>, CliError> {
let db = GraphDb::open(dir)?;
Ok(db.suggest_rules())
}
pub fn format_suggest(suggestions: &[RuleSuggestion]) -> String {
let mut buf = String::new();
if suggestions.is_empty() {
let _ = writeln!(
buf,
"no rule suggestions (database may be empty or rules already cover all patterns)"
);
return buf;
}
let _ = writeln!(buf, "== rule suggestions ({}) ==", suggestions.len());
for (i, s) in suggestions.iter().enumerate() {
let _ = writeln!(buf);
let _ = writeln!(buf, "[{}] {}", i + 1, s.def.name);
let _ = writeln!(
buf,
" {} → {} via {:?}",
s.def.src_label, s.def.dst_label, s.def.predicate
);
let _ = writeln!(buf, " est_edges : ~{}", s.est_edges);
let _ = writeln!(buf, " rationale : {}", s.rationale);
if !s.examples.is_empty() {
let _ = writeln!(buf, " examples :");
for (src, dst, score) in &s.examples {
let _ = writeln!(buf, " {src} → {dst} score={score:.4}");
}
}
let _ = writeln!(buf, " predicate : {:?}", s.def.predicate);
let _ = writeln!(
buf,
" to apply : POST /rules or db.create_rule(suggestion.def)"
);
}
buf
}
fn fmt_value(v: &Value) -> String {
match v {
Value::Int(i) => i.to_string(),
Value::Float(f) => {
let s = format!("{f}");
if s.contains('.') || s.contains('e') || s.contains('E') {
s
} else {
format!("{s}.0")
}
}
Value::Str(s) => s.clone(),
Value::Bool(b) => b.to_string(),
Value::List(xs) => {
let inner: Vec<String> = xs.iter().map(fmt_value).collect();
format!("[{}]", inner.join(", "))
}
Value::Map(m) => {
let inner: Vec<String> = m
.iter()
.map(|(k, v)| format!("{k}: {}", fmt_value(v)))
.collect();
format!("{{{}}}", inner.join(", "))
}
}
}
fn fmt_cell(cell: Option<&Value>) -> String {
match cell {
None => "null".into(),
Some(v) => fmt_value(v),
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeSet;
use std::net::SocketAddr;
use std::path::PathBuf;
fn tmp(name: &str) -> PathBuf {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("clock")
.as_nanos();
let d = std::env::temp_dir().join(format!(
"graphdb-cli-{}-{}-{}",
name,
std::process::id(),
nanos
));
let _ = std::fs::remove_dir_all(&d);
d
}
fn directed_pairs(db: &SharedDb, etype: &str) -> BTreeSet<(String, String)> {
let g = db.read();
let mut out = BTreeSet::new();
for i in 1..=N_ORGS {
let src = format!("org-{i:02}");
if let Ok(nbrs) = g.neighbors(&src, etype, core_api::Direction::Out) {
for dst in nbrs {
out.insert((src.clone(), dst));
}
}
}
for i in 1..=N_PEOPLE {
let src = format!("person-{i:02}");
if let Ok(nbrs) = g.neighbors(&src, etype, core_api::Direction::Out) {
for dst in nbrs {
out.insert((src.clone(), dst));
}
}
}
out
}
fn assert_weight(db: &SharedDb, a: &str, b: &str, rule: &str, want: f64) {
let hits: Vec<_> = db
.read()
.explain(a, b)
.expect("explain")
.into_iter()
.filter(|e| e.rule == rule && e.src_key == a && e.dst_key == b)
.collect();
assert_eq!(hits.len(), 1, "explain {a}/{b} rule={rule}: {hits:?}");
let got = hits[0].weight.expect("weighted");
assert!(
(got - want).abs() < 1e-12,
"{rule} {a}→{b}: got {got} want {want}"
);
}
fn haversine_km(lat1: f64, lon1: f64, lat2: f64, lon2: f64) -> f64 {
const R: f64 = 6371.0088;
let phi1 = lat1.to_radians();
let phi2 = lat2.to_radians();
let dphi = (lat2 - lat1).to_radians();
let dlam = (lon2 - lon1).to_radians();
let a = ((dphi / 2.0).sin().powi(2) + phi1.cos() * phi2.cos() * (dlam / 2.0).sin().powi(2))
.clamp(0.0, 1.0);
let c = 2.0 * a.sqrt().atan2((1.0 - a).sqrt());
R * c
}
fn default_bind() -> SocketAddr {
SocketAddr::from(([127, 0, 0, 1], 8080))
}
#[test]
fn parse_args_table() {
struct Case {
args: &'static [&'static str],
check: fn(Result<Command, String>),
}
let cases = [
Case {
args: &[],
check: |r| match r {
Ok(Command::Help) => {}
other => panic!("no-args → Help, got {other:?}"),
},
},
Case {
args: &["--help"],
check: |r| match r {
Ok(Command::Help) => {}
other => panic!("--help → Help, got {other:?}"),
},
},
Case {
args: &["-h"],
check: |r| match r {
Ok(Command::Help) => {}
other => panic!("-h → Help, got {other:?}"),
},
},
Case {
args: &["serve", "/tmp/demo-db"],
check: |r| match r {
Ok(Command::Serve {
db_dir,
addr,
ui,
demo_if_empty,
token,
role_tokens,
snapshot_every,
}) => {
assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
assert_eq!(addr, default_bind());
assert_eq!(ui, super::ServeUi::Embedded);
assert!(!demo_if_empty);
assert_eq!(token, None);
assert!(role_tokens.is_empty());
assert_eq!(snapshot_every, None);
}
other => panic!("serve <dir> → Serve default addr, got {other:?}"),
},
},
Case {
args: &["serve", "/tmp/demo-db", "--addr", "127.0.0.1:8080"],
check: |r| match r {
Ok(Command::Serve {
db_dir,
addr,
ui,
demo_if_empty,
token,
role_tokens,
snapshot_every,
}) => {
assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
assert_eq!(
addr,
"127.0.0.1:8080".parse::<std::net::SocketAddr>().unwrap()
);
assert_eq!(ui, super::ServeUi::Embedded);
assert!(!demo_if_empty);
assert_eq!(token, None);
assert!(role_tokens.is_empty());
assert_eq!(snapshot_every, None);
}
other => panic!("serve --addr after dir, got {other:?}"),
},
},
Case {
args: &["serve", "/tmp/demo-db", "--addr=127.0.0.1:9090"],
check: |r| match r {
Ok(Command::Serve {
db_dir,
addr,
ui,
demo_if_empty,
token,
role_tokens,
snapshot_every,
}) => {
assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
assert_eq!(
addr,
"127.0.0.1:9090".parse::<std::net::SocketAddr>().unwrap()
);
assert_eq!(ui, super::ServeUi::Embedded);
assert!(!demo_if_empty);
assert_eq!(token, None);
let _ = role_tokens; assert_eq!(snapshot_every, None);
}
other => panic!("serve --addr=VALUE, got {other:?}"),
},
},
Case {
args: &["mcp", "/tmp/demo-db"],
check: |r| match r {
Ok(Command::Mcp { db_dir }) => {
assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
}
other => panic!("mcp <dir>, got {other:?}"),
},
},
Case {
args: &["stats", "/tmp/demo-db"],
check: |r| match r {
Ok(Command::Stats { db_dir }) => {
assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
}
other => panic!("stats <dir>, got {other:?}"),
},
},
Case {
args: &["demo", "/tmp/demo-db"],
check: |r| match r {
Ok(Command::Demo { db_dir }) => {
assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
}
other => panic!("demo <dir>, got {other:?}"),
},
},
Case {
args: &["serve"],
check: |r| {
let e = r.expect_err("serve without dir");
assert!(
e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
"missing-dir error should mention dir, got {e}"
);
},
},
Case {
args: &["mcp"],
check: |r| {
let e = r.expect_err("mcp without dir");
assert!(
e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
"missing-dir error should mention dir, got {e}"
);
},
},
Case {
args: &["stats"],
check: |r| {
let e = r.expect_err("stats without dir");
assert!(
e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
"missing-dir error should mention dir, got {e}"
);
},
},
Case {
args: &["demo"],
check: |r| {
let e = r.expect_err("demo without dir");
assert!(
e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
"missing-dir error should mention dir, got {e}"
);
},
},
Case {
args: &["serve", "/tmp/demo-db", "--addr"],
check: |r| {
let e = r.expect_err("--addr missing value");
assert!(
e.to_lowercase().contains("addr"),
"--addr missing value should mention addr, got {e}"
);
},
},
Case {
args: &["serve", "/tmp/demo-db", "--addr", "not-an-addr"],
check: |r| {
let e = r.expect_err("invalid addr");
assert!(
e.to_lowercase().contains("addr") || e.to_lowercase().contains("address"),
"invalid addr should mention address, got {e}"
);
},
},
Case {
args: &["frobnicate", "/tmp/demo-db"],
check: |r| {
let e = r.expect_err("unknown command");
assert!(
e.to_lowercase().contains("unknown")
|| e.to_lowercase().contains("frobnicate"),
"unknown command should name it, got {e}"
);
},
},
Case {
args: &["serve", "/tmp/demo-db", "--ui", "/tmp/ui-dist"],
check: |r| match r {
Ok(Command::Serve { ui, .. }) => {
assert_eq!(
ui,
super::ServeUi::Filesystem(PathBuf::from("/tmp/ui-dist"))
);
}
other => panic!("serve --ui <dir>, got {other:?}"),
},
},
Case {
args: &["serve", "/tmp/demo-db", "--ui=/tmp/ui-eq"],
check: |r| match r {
Ok(Command::Serve { ui, .. }) => {
assert_eq!(ui, super::ServeUi::Filesystem(PathBuf::from("/tmp/ui-eq")));
}
other => panic!("serve --ui=VALUE, got {other:?}"),
},
},
Case {
args: &["serve", "/tmp/demo-db", "--ui"],
check: |r| {
let e = r.expect_err("--ui missing value");
assert!(
e.to_lowercase().contains("ui"),
"--ui missing value should mention ui, got {e}"
);
},
},
Case {
args: &["serve", "/tmp/demo-db", "--no-ui"],
check: |r| match r {
Ok(Command::Serve { ui, .. }) => {
assert_eq!(ui, super::ServeUi::None);
}
other => panic!("serve --no-ui, got {other:?}"),
},
},
Case {
args: &["serve", "/tmp/demo-db", "--ui", "/tmp/x", "--no-ui"],
check: |r| {
let e = r.expect_err("combine --ui and --no-ui");
assert!(
e.contains("--ui") && e.contains("--no-ui"),
"conflict should name both flags, got {e}"
);
},
},
Case {
args: &["serve", "/tmp/demo-db", "extra"],
check: |r| {
let e = r.expect_err("extra positional");
assert!(
e.to_lowercase().contains("unexpected")
|| e.to_lowercase().contains("extra"),
"extra arg should be rejected, got {e}"
);
},
},
Case {
args: &[
"serve",
"/data",
"--addr",
"0.0.0.0:8080",
"--demo-if-empty",
],
check: |r| match r {
Ok(Command::Serve {
db_dir,
addr,
demo_if_empty,
ui,
token,
snapshot_every,
..
}) => {
assert_eq!(db_dir, PathBuf::from("/data"));
assert_eq!(
addr,
"0.0.0.0:8080".parse::<std::net::SocketAddr>().unwrap()
);
assert!(demo_if_empty);
assert_eq!(ui, super::ServeUi::Embedded);
assert_eq!(token, None);
assert_eq!(snapshot_every, None);
}
other => panic!("serve --demo-if-empty docker default, got {other:?}"),
},
},
];
for case in &cases {
(case.check)(parse_args(case.args));
}
}
#[test]
fn serve_default_addr_is_loopback_8080() {
match parse_args(&["serve", "/tmp/db"]).unwrap() {
Command::Serve { addr, .. } => {
assert_eq!(
addr,
"127.0.0.1:8080".parse::<std::net::SocketAddr>().unwrap()
);
}
other => panic!("{other:?}"),
}
}
#[test]
fn serve_snapshot_every_parses_seconds() {
match parse_args(&["serve", "/tmp/db", "--snapshot-every", "30"]).unwrap() {
Command::Serve { snapshot_every, .. } => {
assert_eq!(snapshot_every, Some(Duration::from_secs(30)));
}
other => panic!("{other:?}"),
}
match parse_args(&["serve", "/tmp/db", "--snapshot-every=5"]).unwrap() {
Command::Serve { snapshot_every, .. } => {
assert_eq!(snapshot_every, Some(Duration::from_secs(5)));
}
other => panic!("{other:?}"),
}
match parse_args(&["serve", "/tmp/db"]).unwrap() {
Command::Serve { snapshot_every, .. } => {
assert_eq!(snapshot_every, None);
}
other => panic!("{other:?}"),
}
let err = parse_args(&["serve", "/tmp/db", "--snapshot-every"]).unwrap_err();
assert!(
err.contains("snapshot-every"),
"missing value should name the flag, got {err}"
);
let err = parse_args(&["serve", "/tmp/db", "--snapshot-every", "0"]).unwrap_err();
assert!(
err.contains("snapshot-every"),
"zero should be rejected, got {err}"
);
let err = parse_args(&["serve", "/tmp/db", "--snapshot-every", "nope"]).unwrap_err();
assert!(
err.contains("snapshot-every"),
"invalid value should name the flag, got {err}"
);
}
#[test]
fn serve_token_flag_and_non_loopback_without_token_is_parsed() {
match parse_args(&[
"serve",
"/tmp/db",
"--addr",
"0.0.0.0:8080",
"--token",
"s3cret",
])
.unwrap()
{
Command::Serve { token, addr, .. } => {
assert_eq!(token.as_deref(), Some("s3cret"));
assert_eq!(addr.ip().to_string(), "0.0.0.0");
}
other => panic!("{other:?}"),
}
}
#[test]
fn parse_snapshot_and_query() {
match parse_args(&["snapshot", "/tmp/db"]).unwrap() {
Command::Snapshot { keep_wal, .. } => assert!(!keep_wal),
other => panic!("{other:?}"),
}
match parse_args(&["snapshot", "/tmp/db", "--keep-wal"]).unwrap() {
Command::Snapshot { keep_wal, .. } => assert!(keep_wal),
other => panic!("{other:?}"),
}
match parse_args(&["query", "/tmp/db", "MATCH (n) RETURN n LIMIT 1"]).unwrap() {
Command::Query { cypher, .. } => assert!(cypher.contains("MATCH")),
other => panic!("{other:?}"),
}
match parse_args(&["query", "/tmp/db", "MATCH", "(n)", "RETURN", "n"]).unwrap() {
Command::Query { cypher, .. } => assert_eq!(cypher, "MATCH (n) RETURN n"),
other => panic!("{other:?}"),
}
match parse_args(&["query", "/tmp/db", "--query", "MATCH (n) RETURN n"]).unwrap() {
Command::Query { cypher, .. } => assert_eq!(cypher, "MATCH (n) RETURN n"),
other => panic!("{other:?}"),
}
let text = usage();
assert!(
text.contains("query"),
"usage should mention query, got:\n{text}"
);
assert!(
text.contains("snapshot"),
"usage should mention snapshot, got:\n{text}"
);
}
#[test]
fn usage_lists_every_subcommand() {
let text = usage();
for word in [
"serve",
"mcp",
"stats",
"demo",
"query",
"snapshot",
"--keep-wal",
"mushroomdb",
"--ui",
"--no-ui",
"--demo-if-empty",
"--token",
"--snapshot-every",
] {
assert!(
text.contains(word),
"usage should mention {word}, got:\n{text}"
);
}
}
#[test]
fn validate_ui_dir_requires_index_html() {
let missing = tmp("ui-missing");
let err = super::validate_ui_dir(&missing).expect_err("missing dir");
assert!(
err.contains("does not exist"),
"missing dir error, got {err}"
);
let empty = tmp("ui-empty");
std::fs::create_dir_all(&empty).unwrap();
let err = super::validate_ui_dir(&empty).expect_err("no index");
assert!(
err.contains("index.html"),
"missing index.html error, got {err}"
);
let ok = tmp("ui-ok");
std::fs::create_dir_all(&ok).unwrap();
std::fs::write(ok.join("index.html"), "<!doctype html>").unwrap();
let got = super::validate_ui_dir(&ok).expect("valid ui dir");
assert_eq!(got, ok);
}
#[test]
fn maybe_run_demo_if_empty_seeds_then_skips() {
let dir = tmp("boot-empty");
let first = super::maybe_run_demo_if_empty(&dir)
.expect("empty dir demos")
.expect("Some(DemoOutcome)");
assert_eq!(first.stats.nodes_live, 60);
let db = SharedDb::open(&dir).expect("reopen");
assert!(db.read().has_node("person-01"));
let second = super::maybe_run_demo_if_empty(&dir).expect("non-empty is ok");
assert!(
second.is_none(),
"second boot must not re-demo a populated volume"
);
let occupied = tmp("boot-occupied");
std::fs::create_dir_all(&occupied).unwrap();
std::fs::write(occupied.join("keep-me"), b"x").unwrap();
let skipped = super::maybe_run_demo_if_empty(&occupied).expect("occupied skip");
assert!(skipped.is_none());
assert_eq!(
std::fs::read(occupied.join("keep-me")).unwrap(),
b"x",
"existing volume contents must be untouched"
);
}
#[test]
fn demo_builder_is_deterministic_and_refuses_second_run() {
let dir = tmp("demo");
let out = run_demo(&dir).expect("first demo run");
assert_eq!(
out.stats.nodes_live, 60,
"10 orgs + 20 projects + 30 people"
);
assert_eq!(out.stats.nodes_tombstoned, 0);
assert_eq!(out.stats.edges, 334);
assert_eq!(
out.stats.rules.len(),
7,
"3 auto-FK + overlap + numeric + geo + vector"
);
let fit = out
.stats
.rules
.iter()
.find(|r| r.name == "skill_fit")
.expect("skill_fit");
assert_eq!(fit.edges, 90, "30 people × 3 FIT edges");
let founded = out
.stats
.rules
.iter()
.find(|r| r.name == "founded_within")
.expect("founded_within");
assert_eq!(founded.edges, 34);
let nearby = out
.stats
.rules
.iter()
.find(|r| r.name == "nearby_office")
.expect("nearby_office");
assert_eq!(nearby.edges, 16);
let similar = out
.stats
.rules
.iter()
.find(|r| r.name == "similar_interests")
.expect("similar_interests");
assert_eq!(similar.edges, 114);
let mut names: Vec<&str> = out.stats.rules.iter().map(|r| r.name.as_str()).collect();
names.sort_unstable();
assert_eq!(
names,
vec![
"auto_fk_person_org_id",
"auto_fk_person_project_id",
"auto_fk_project_org_id",
"founded_within",
"nearby_office",
"similar_interests",
"skill_fit",
]
);
let mut auto = out.auto_fk_rules.clone();
auto.sort();
assert_eq!(
auto,
vec![
"auto_fk_person_org_id".to_string(),
"auto_fk_person_project_id".to_string(),
"auto_fk_project_org_id".to_string(),
]
);
assert!(
!out.sample_result.is_empty(),
"sample Cypher query must return rows"
);
assert!(
out.sample_query.contains("ORDER BY score DESC"),
"sample query must rank by score, got {}",
out.sample_query
);
let scores: Vec<f64> = (0..out.sample_result.len())
.map(|i| match out.sample_result.get(i, "score") {
Some(Value::Float(f)) => *f,
other => panic!("score col should be Float, got {other:?}"),
})
.collect();
let distinct: std::collections::BTreeSet<u64> =
scores.iter().map(|s| s.to_bits()).collect();
assert!(
distinct.len() >= 2,
"sample results must be visibly ranked, got {scores:?}"
);
for w in scores.windows(2) {
assert!(
w[0] >= w[1],
"scores must be non-increasing, got {scores:?}"
);
}
assert!(
!out.explanations.is_empty(),
"explain(person-01, proj-01) must find the derived edges"
);
let db = SharedDb::open(&dir).expect("reopen demo");
assert_eq!(
directed_pairs(&db, "FOUNDED_WITHIN"),
[
("org-01", "org-02"),
("org-01", "org-03"),
("org-02", "org-01"),
("org-02", "org-03"),
("org-02", "org-04"),
("org-03", "org-01"),
("org-03", "org-02"),
("org-03", "org-04"),
("org-03", "org-05"),
("org-04", "org-02"),
("org-04", "org-03"),
("org-04", "org-05"),
("org-04", "org-06"),
("org-05", "org-03"),
("org-05", "org-04"),
("org-05", "org-06"),
("org-05", "org-07"),
("org-06", "org-04"),
("org-06", "org-05"),
("org-06", "org-07"),
("org-06", "org-08"),
("org-07", "org-05"),
("org-07", "org-06"),
("org-07", "org-08"),
("org-07", "org-09"),
("org-08", "org-06"),
("org-08", "org-07"),
("org-08", "org-09"),
("org-08", "org-10"),
("org-09", "org-07"),
("org-09", "org-08"),
("org-09", "org-10"),
("org-10", "org-08"),
("org-10", "org-09"),
]
.into_iter()
.map(|(a, b)| (a.to_string(), b.to_string()))
.collect::<BTreeSet<_>>()
);
assert_eq!(
directed_pairs(&db, "NEARBY_OFFICE"),
[
("org-01", "org-07"),
("org-01", "org-10"),
("org-02", "org-09"),
("org-03", "org-08"),
("org-04", "org-05"),
("org-04", "org-06"),
("org-05", "org-04"),
("org-05", "org-06"),
("org-06", "org-04"),
("org-06", "org-05"),
("org-07", "org-01"),
("org-07", "org-10"),
("org-08", "org-03"),
("org-09", "org-02"),
("org-10", "org-01"),
("org-10", "org-07"),
]
.into_iter()
.map(|(a, b)| (a.to_string(), b.to_string()))
.collect::<BTreeSet<_>>()
);
assert_weight(&db, "org-01", "org-02", "founded_within", 0.5);
let nyc_jc = 1.0 - haversine_km(40.7128, -74.0060, 40.7178, -74.0431) / 50.0;
assert_weight(&db, "org-01", "org-07", "nearby_office", nyc_jc);
assert_weight(&db, "person-01", "person-11", "similar_interests", 1.0);
assert_weight(&db, "person-01", "person-09", "similar_interests", 0.8);
let err = run_demo(&dir).expect_err("second run into the same dir");
let msg = err.to_string().to_lowercase();
assert!(
msg.contains("not empty") || msg.contains("non-empty") || msg.contains("non empty"),
"refuse message must mention non-empty dir, got {err}"
);
assert!(
msg.contains("hidden"),
"refuse message must mention hidden files, got {err}"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn run_snapshot_writes_snapshot_bin() {
let dir = tmp("snapshot-cli");
{
let mut db = GraphDb::open(&dir).expect("open");
db.insert_node("Person", "alice", vec![]).expect("insert");
}
assert!(
!dir.join("snapshot.bin").exists(),
"GraphDb Drop must not snapshot"
);
let out = run_snapshot(&dir, false, false, None).expect("snapshot");
assert!(
dir.join("snapshot.bin").is_file(),
"run_snapshot must write snapshot.bin"
);
assert!(
out.contains("snapshot.bin"),
"snapshot output should mention snapshot.bin, got {out}"
);
let db = GraphDb::open(&dir).expect("reopen");
assert!(db.has_node("alice"), "reopen after snapshot must recover");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn run_query_formats_like_asof() {
let dir = tmp("query-cli");
{
let mut db = GraphDb::open(&dir).expect("open");
db.insert_node(
"Person",
"alice",
vec![("id".into(), Value::Str("alice".into()))],
)
.expect("insert");
}
let out = run_query(&dir, "MATCH (n:Person) RETURN n.id AS id").expect("query");
assert!(out.contains("columns:"), "got {out}");
assert!(out.contains("id=alice"), "got {out}");
let _ = run_query(&dir, "CREATE (n:Person {id: 'bob'})").expect("write");
let db = GraphDb::open(&dir).expect("reopen");
assert!(db.has_node("bob"), "query_write must persist CREATE");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn format_stats_contains_counts() {
let dir = tmp("stats-smoke");
let out = run_demo(&dir).expect("demo for stats smoke");
let text = format_stats(&out.stats);
assert!(
text.contains("60"),
"stats output should include live node count, got:\n{text}"
);
assert!(
text.contains("334"),
"stats output should include edge count, got:\n{text}"
);
assert!(
text.to_lowercase().contains("node"),
"stats output should mention nodes, got:\n{text}"
);
assert!(
text.to_lowercase().contains("edge"),
"stats output should mention edges, got:\n{text}"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn parse_backup_round_trip() {
let r = parse_args(&["backup", "/db/dir", "/backup/dest"]);
match r {
Ok(Command::Backup { db_dir, dest }) => {
assert_eq!(db_dir, PathBuf::from("/db/dir"));
assert_eq!(dest, PathBuf::from("/backup/dest"));
}
other => panic!("backup parse, got {other:?}"),
}
}
#[test]
fn parse_backup_missing_dest_errors() {
let r = parse_args(&["backup", "/db/dir"]);
assert!(r.is_err(), "backup without <dest> should error");
let e = r.unwrap_err();
assert!(
e.to_lowercase().contains("dest"),
"error should mention dest, got: {e}"
);
}
#[test]
fn parse_export_defaults_to_jsonl() {
let r = parse_args(&["export", "/db/dir", "/export/dest"]);
match r {
Ok(Command::Export { format, .. }) => {
assert_eq!(format, ExportFormat::Jsonl);
}
other => panic!("export parse, got {other:?}"),
}
}
#[test]
fn parse_export_parquet_flag() {
let r = parse_args(&["export", "/db/dir", "/export/dest", "--format", "parquet"]);
match r {
Ok(Command::Export { format, .. }) => {
assert_eq!(format, ExportFormat::Parquet);
}
other => panic!("export --format parquet parse, got {other:?}"),
}
}
#[test]
fn parse_export_parquet_flag_eq() {
let r = parse_args(&["export", "/db/dir", "/dest", "--format=parquet"]);
match r {
Ok(Command::Export { format, .. }) => {
assert_eq!(format, ExportFormat::Parquet);
}
other => panic!("export --format=parquet parse, got {other:?}"),
}
}
#[test]
fn run_backup_cli_produces_verified_report() {
let src = tmp("cli-backup-src");
let dst = tmp("cli-backup-dst");
let _ = run_demo(&src).expect("demo");
let report = run_backup(&src, &dst).expect("run_backup");
assert!(report.verified, "backup must be verified");
assert!(!report.files.is_empty());
assert!(report.bytes > 0);
let _ = std::fs::remove_dir_all(&src);
let _ = std::fs::remove_dir_all(&dst);
}
#[test]
fn run_export_jsonl_two_runs_byte_identical() {
let src = tmp("cli-export-src");
let dst1 = tmp("cli-export-dst1");
let dst2 = tmp("cli-export-dst2");
let _ = run_demo(&src).expect("demo");
run_export(&src, &dst1, &ExportFormat::Jsonl).expect("first export");
run_export(&src, &dst2, &ExportFormat::Jsonl).expect("second export");
for filename in &["nodes.jsonl", "edges.jsonl", "rules.jsonl"] {
let f1 = std::fs::read(dst1.join(filename)).expect("read first");
let f2 = std::fs::read(dst2.join(filename)).expect("read second");
assert_eq!(
f1, f2,
"{filename} must be byte-identical across two export runs"
);
}
let _ = std::fs::remove_dir_all(&src);
let _ = std::fs::remove_dir_all(&dst1);
let _ = std::fs::remove_dir_all(&dst2);
}
#[test]
fn run_export_jsonl_nodes_are_sorted() {
let src = tmp("cli-export-sorted");
let dst = tmp("cli-export-sorted-dst");
let _ = run_demo(&src).expect("demo");
run_export(&src, &dst, &ExportFormat::Jsonl).expect("export");
let content = std::fs::read_to_string(dst.join("nodes.jsonl")).expect("read nodes");
let keys: Vec<String> = content
.lines()
.filter(|l| !l.is_empty())
.map(|l| {
let v: serde_json::Value = serde_json::from_str(l).expect("parse line");
v["key"].as_str().unwrap_or("").to_string()
})
.collect();
let mut sorted = keys.clone();
sorted.sort();
assert_eq!(keys, sorted, "nodes.jsonl must be sorted by key");
let _ = std::fs::remove_dir_all(&src);
let _ = std::fs::remove_dir_all(&dst);
}
#[test]
fn run_export_jsonl_derived_edges_have_rule() {
let src = tmp("cli-export-derived");
let dst = tmp("cli-export-derived-dst");
let _ = run_demo(&src).expect("demo");
run_export(&src, &dst, &ExportFormat::Jsonl).expect("export");
let content = std::fs::read_to_string(dst.join("edges.jsonl")).expect("read edges");
let derived_lines: Vec<serde_json::Value> = content
.lines()
.filter(|l| !l.is_empty())
.map(|l| serde_json::from_str(l).expect("parse line"))
.filter(|v: &serde_json::Value| v["derived"].as_bool().unwrap_or(false))
.collect();
assert!(
!derived_lines.is_empty(),
"demo store should have derived edges"
);
for edge in &derived_lines {
assert!(
!edge["rule"].is_null(),
"derived edge must have non-null rule: {edge}"
);
}
let _ = std::fs::remove_dir_all(&src);
let _ = std::fs::remove_dir_all(&dst);
}
#[test]
fn run_export_parquet_produces_files() {
let src = tmp("cli-export-parq-src");
let dst = tmp("cli-export-parq-dst");
let _ = run_demo(&src).expect("demo");
run_export(&src, &dst, &ExportFormat::Parquet).expect("parquet export");
assert!(
dst.join("nodes.parquet").exists(),
"nodes.parquet must exist"
);
assert!(
dst.join("edges.parquet").exists(),
"edges.parquet must exist"
);
assert!(
dst.join("rules.parquet").exists(),
"rules.parquet must exist"
);
for f in &["nodes.parquet", "edges.parquet", "rules.parquet"] {
let meta = std::fs::metadata(dst.join(f)).expect("metadata");
assert!(meta.len() > 0, "{f} must be non-empty");
}
let _ = std::fs::remove_dir_all(&src);
let _ = std::fs::remove_dir_all(&dst);
}
#[test]
fn run_export_jsonl_nan_float_becomes_null() {
use core_api::{GraphDb, Value};
let src = tmp("cli-export-nan-src");
let dst = tmp("cli-export-nan-dst");
{
let mut db = GraphDb::open(&src).unwrap();
db.insert_node(
"Sensor",
"s1",
vec![
("nan_val".into(), Value::Float(f64::NAN)),
("pos_inf".into(), Value::Float(f64::INFINITY)),
("neg_inf".into(), Value::Float(f64::NEG_INFINITY)),
("normal".into(), Value::Float(1.5)),
],
)
.unwrap();
}
run_export(&src, &dst, &ExportFormat::Jsonl).expect("export with NaN must succeed");
let content =
std::fs::read_to_string(dst.join("nodes.jsonl")).expect("nodes.jsonl missing");
let row: serde_json::Value =
serde_json::from_str(content.lines().next().unwrap()).expect("valid json line");
assert_eq!(
row["nan_val"],
serde_json::Value::Null,
"NaN must export as null"
);
assert_eq!(
row["pos_inf"],
serde_json::Value::Null,
"+Inf must export as null"
);
assert_eq!(
row["neg_inf"],
serde_json::Value::Null,
"-Inf must export as null"
);
assert_eq!(
row["normal"],
serde_json::json!(1.5),
"normal float roundtrips"
);
let _ = std::fs::remove_dir_all(&src);
let _ = std::fs::remove_dir_all(&dst);
}
}