crates/graphar-flight/src/bin/graphar-flight.rs
//! `graphar-flight` — the standalone Flight SQL server over a FalkorDB graph.
//!
//! ```text
//! graphar-flight --falkor redis://127.0.0.1:6379 --graph G --listen 0.0.0.0:50056 [--token T] [--basic USER:PASS]
//! [--table NAME=<cypher>]… [--tls-cert F --tls-key F [--tls-client-ca F]]
//! graphar-flight --unit | --install | --uninstall | --start | --stop | --status [--system] [--env K=V]…
//! ```
//!
//! Until 2026-09-10 this crate had a library and an example and NO binary: the
//! only way to serve Flight SQL over a graph was to embed the service (korp does),
//! and the only way to try it was four side-by-side checkouts. `skade-flight`
//! builds and serves from one clone; this is the same door for the graph side.
//!
//! Token ladder (nornir-service, the ONE ladder): `--token` → `$GRAPHAR_FLIGHT_TOKEN`
//! → `<home>/.graphar-flight/token`. Empty everywhere ⇒ OPEN server (said loudly).
//!
//! Tables: every `--table NAME=<cypher>` is advertised to a metadata walk (Power
//! BI, ADBC) as a table whose `SELECT * FROM NAME` runs that Cypher. With none
//! given, two generic tables are advertised — `graph_nodes` (`_gar_id`, `name`) and
//! `graph_edges` (`src`, `dst`, `rel`) — the same pair korp's host serves, so a
//! client sees something to click on an unconfigured graph.
//!
//! Lifecycle verbs render and install the systemd unit through skidbladnir
//! (LAW: a nordisk service installs through its own door, never a hand-copied
//! binary or a hand-written drop-in).
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::{Context, Result, bail};
use arrow_schema::{DataType, Field, Schema};
use graphar_flight::{SchemaRegistry, ServerOptions, run_server_with};
const USAGE: &str = "\
graphar-flight — Apache Arrow Flight SQL over a FalkorDB graph (Cypher in, Arrow out)
--falkor URL FalkorDB/redis URL (env GRAPHAR_FLIGHT_FALKOR; default redis://127.0.0.1:6379)
--graph NAME the graph to serve (env GRAPHAR_FLIGHT_GRAPH; default `graph`)
--listen ADDR bind address (env GRAPHAR_FLIGHT_ADDR; default 127.0.0.1:50056)
--token T bearer token (env GRAPHAR_FLIGHT_TOKEN, file ~/.graphar-flight/token); empty = OPEN
--basic USER:PASS Basic credential the Handshake exchanges for the token (env GRAPHAR_FLIGHT_BASIC)
--table NAME=CYPHER advertise NAME as a table backed by CYPHER (repeatable; columns typed Utf8
except `_gar_id`/`src`/`dst`, Int64). None given ⇒ graph_nodes + graph_edges.
--tls-cert F --tls-key F [--tls-client-ca F] TLS / mTLS on the listener
--unit print the systemd unit --install writes
--install render + install + enable + start the unit via skidbladnir
--uninstall stop, disable, remove the unit (data root untouched unless --purge)
--start/--stop/--status
--system system scope (User=graphar-flight); default is --user scope
--env K=V (with --install, repeatable) operator env rendered into the unit
";
/// What the command line asked for — pure, so the parse is testable without a
/// socket, a graph or a systemd. Every default is stated here once.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Cli {
pub falkor: String,
pub graph: String,
pub listen: String,
pub token_flag: Option<String>,
pub basic: Option<(String, String)>,
/// `(name, cypher)` in argv order.
pub tables: Vec<(String, String)>,
pub tls: Option<(String, String, Option<String>)>,
pub lifecycle: Option<String>,
pub system: bool,
}
fn flag(args: &[String], name: &str) -> Option<String> {
let mut it = args.iter();
while let Some(a) = it.next() {
if a == name {
return it.next().cloned();
}
if let Some(v) = a.strip_prefix(&format!("{name}=")) {
return Some(v.to_string());
}
}
None
}
/// Every occurrence of a repeatable `--name VALUE` / `--name=VALUE`, in order.
fn flags(args: &[String], name: &str) -> Vec<String> {
let mut out = Vec::new();
let mut it = args.iter();
while let Some(a) = it.next() {
if a == name {
if let Some(v) = it.next() {
out.push(v.clone());
}
} else if let Some(v) = a.strip_prefix(&format!("{name}=")) {
out.push(v.to_string());
}
}
out
}
const LIFECYCLE: [&str; 6] = ["--unit", "--install", "--uninstall", "--start", "--stop", "--status"];
/// Parse argv with `env` standing in for the process environment (so a test can
/// pass its own). Flags win over env, env over the documented defaults.
pub fn parse(args: &[String], env: &dyn Fn(&str) -> Option<String>) -> Result<Cli> {
let pick = |name: &str, var: &str, default: &str| {
flag(args, name).or_else(|| env(var)).unwrap_or_else(|| default.to_string())
};
let basic = match flag(args, "--basic").or_else(|| env("GRAPHAR_FLIGHT_BASIC")) {
Some(b) => {
let Some((u, p)) = b.split_once(':') else { bail!("--basic wants USER:PASS") };
Some((u.to_string(), p.to_string()))
}
None => None,
};
let mut tables = Vec::new();
for t in flags(args, "--table") {
let Some((name, cypher)) = t.split_once('=') else {
bail!("--table wants NAME=<cypher>, got `{t}`");
};
if name.trim().is_empty() || cypher.trim().is_empty() {
bail!("--table wants a non-empty NAME and a non-empty Cypher, got `{t}`");
}
tables.push((name.trim().to_string(), cypher.trim().to_string()));
}
let tls = match (flag(args, "--tls-cert"), flag(args, "--tls-key")) {
(Some(c), Some(k)) => Some((c, k, flag(args, "--tls-client-ca"))),
(None, None) => None,
_ => bail!("--tls-cert and --tls-key go together"),
};
let lifecycle = LIFECYCLE
.iter()
.find(|v| args.iter().any(|a| a == *v))
.map(|v| v.to_string());
Ok(Cli {
falkor: pick("--falkor", "GRAPHAR_FLIGHT_FALKOR", "redis://127.0.0.1:6379"),
graph: pick("--graph", "GRAPHAR_FLIGHT_GRAPH", "graph"),
listen: pick("--listen", "GRAPHAR_FLIGHT_ADDR", "127.0.0.1:50056"),
token_flag: flag(args, "--token"),
basic,
tables,
tls,
lifecycle,
system: args.iter().any(|a| a == "--system"),
})
}
/// The tables a metadata walk sees: the operator's `--table`s, or the two
/// generic ones when none were given. Columns are `Int64` for the id columns a
/// graph export carries (`_gar_id`, `src`, `dst`) and `Utf8` otherwise — the
/// same rule `SchemaRegistry::register_vertex_query` applies.
pub fn build_registry(tables: &[(String, String)]) -> Arc<SchemaRegistry> {
let registry = Arc::new(SchemaRegistry::new());
if tables.is_empty() {
registry.register_named(
"graph_nodes",
"MATCH (n) RETURN n._gar_id AS _gar_id, n.name AS name",
Arc::new(Schema::new(vec![
Field::new("_gar_id", DataType::Int64, false),
Field::new("name", DataType::Utf8, true),
])),
);
registry.register_named(
"graph_edges",
"MATCH (a)-[r]->(b) RETURN a._gar_id AS src, b._gar_id AS dst, type(r) AS rel",
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("rel", DataType::Utf8, true),
])),
);
return registry;
}
for (name, cypher) in tables {
let fields: Vec<Field> = return_columns(cypher)
.into_iter()
.map(|c| {
let int = matches!(c.as_str(), "_gar_id" | "src" | "dst");
Field::new(c, if int { DataType::Int64 } else { DataType::Utf8 }, !int)
})
.collect();
registry.register_named(name.clone(), cypher.clone(), Arc::new(Schema::new(fields)));
}
registry
}
/// The column names a `RETURN a AS x, b.c, d` clause produces — the alias when
/// there is one, the expression text otherwise. Pure text; a Cypher with no
/// RETURN yields no columns (the engine will reject it at query time).
pub fn return_columns(cypher: &str) -> Vec<String> {
let upper = cypher.to_ascii_uppercase();
let Some(pos) = upper.rfind(" RETURN ") else { return Vec::new() };
let tail = &cypher[pos + " RETURN ".len()..];
// Cut at the first trailing clause; ORDER BY / LIMIT / SKIP do not add columns.
let mut end = tail.len();
for kw in [" ORDER BY ", " LIMIT ", " SKIP "] {
if let Some(i) = tail.to_ascii_uppercase().find(kw) {
end = end.min(i);
}
}
tail[..end]
.split(',')
.map(|expr| {
let expr = expr.trim();
let up = expr.to_ascii_uppercase();
match up.rfind(" AS ") {
Some(i) => expr[i + 4..].trim().to_string(),
None => expr.to_string(),
}
})
.filter(|c| !c.is_empty())
.collect()
}
fn home() -> PathBuf {
std::env::var_os("HOME").map(PathBuf::from).unwrap_or_else(|| PathBuf::from("/"))
}
fn state_home() -> PathBuf {
std::env::var_os("GRAPHAR_FLIGHT_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home().join(".graphar-flight"))
}
fn spec(args: &[String], cli: &Cli) -> Result<skidbladnir::service::ServiceSpec> {
use skidbladnir::service::{AgentMode, ServiceSpec};
use skidbladnir::systemd::{RestartPolicy, Scope};
let exec = std::env::current_exe()
.map(|p| p.to_string_lossy().into_owned())
.unwrap_or_else(|_| "/usr/local/bin/graphar-flight".into());
let mut spec = ServiceSpec::new("graphar-flight", "graphar-flight Flight SQL server", exec)
.scope(if cli.system { Scope::System } else { Scope::User })
.working_dir(state_home().to_string_lossy())
.env("GRAPHAR_FLIGHT_ADDR", &cli.listen)
.env("GRAPHAR_FLIGHT_FALKOR", &cli.falkor)
.env("GRAPHAR_FLIGHT_GRAPH", &cli.graph)
.restart(RestartPolicy::OnFailure)
.agent(AgentMode::Fat);
if cli.system {
spec = spec.user("graphar-flight");
}
for (k, v) in skidbladnir::service::env_args_from_argv(args)? {
spec = spec.operator_env(k, v);
}
match skidbladnir::uninstall::DataRoot::from_env("GRAPHAR_FLIGHT_DATA", state_home()) {
Ok(dr) => spec = spec.data_root(&dr),
Err(e) => eprintln!("graphar-flight: {e} — uninstall would erase nothing; not declaring a root"),
}
Ok(spec)
}
#[tokio::main]
async fn main() -> Result<()> {
let args: Vec<String> = std::env::args().skip(1).collect();
if args.iter().any(|a| a == "-h" || a == "--help") {
print!("{USAGE}");
return Ok(());
}
let cli = parse(&args, &|v| std::env::var(v).ok().filter(|s| !s.is_empty()))?;
// ── lifecycle door ───────────────────────────────────────────────────
if let Some(verb) = cli.lifecycle.as_deref() {
let lc = skidbladnir::service::Lifecycle::new(spec(&args, &cli)?);
match verb {
"--unit" => print!("{}", lc.unit_text()),
"--install" => {
lc.install().context("install")?;
lc.enable().context("enable")?;
lc.start().context("start")?;
println!("graphar-flight: INSTALLED and started ({})", lc.unit());
}
"--uninstall" => {
let r = lc.uninstall().context("uninstall")?;
println!(
"graphar-flight: unit removed = {}, erased {} path(s)",
r.unit_removed,
r.erased.len()
);
}
"--start" => lc.start()?,
"--stop" => lc.stop()?,
"--status" => println!("{}", if lc.status()? { "active" } else { "inactive" }),
_ => unreachable!("LIFECYCLE lists every verb"),
}
return Ok(());
}
// ── serve ────────────────────────────────────────────────────────────
let addr: SocketAddr = cli.listen.parse().with_context(|| format!("--listen {}", cli.listen))?;
let token_file = state_home().join("token");
let (token, whence) = nornir_service::resolve_token_whence(
cli.token_flag.as_deref(),
"GRAPHAR_FLIGHT_TOKEN",
Some(token_file.as_path()),
);
let mut opts = ServerOptions::default();
if token.is_empty() {
eprintln!("graphar-flight: no bearer token ({}) — serving OPEN", whence.describe());
} else {
eprintln!("graphar-flight: bearer from {}", whence.describe());
opts.bearer_token = Some(token);
}
opts.basic_auth = cli.basic.clone();
if let Some((c, k, ca)) = &cli.tls {
opts.tls = Some(match ca {
Some(ca) => graphar_flight::TlsOptions::mutual_from_pem_files(c, k, ca)
.with_context(|| format!("tls {c} {k} {ca}"))?,
None => graphar_flight::TlsOptions::from_pem_files(c, k).with_context(|| format!("tls {c} {k}"))?,
});
}
let registry = build_registry(&cli.tables);
let names = registry.tables().into_iter().map(|(n, _)| n).collect::<Vec<_>>();
eprintln!(
"graphar-flight: graph `{}` at {} — Flight SQL on {addr}; tables {:?}",
cli.graph, cli.falkor, names
);
run_server_with(&cli.falkor, &cli.graph, registry, addr, opts).await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn argv(s: &str) -> Vec<String> {
s.split_whitespace().map(str::to_string).collect()
}
fn no_env(_: &str) -> Option<String> {
None
}
/// **Every documented flag parses, and the ladder is flag > env > default.**
/// RED direction: swap the `or_else` order in `pick` and the second block
/// fails (env would beat the flag); drop a default and the first block fails.
#[test]
fn the_documented_flags_parse_and_flags_beat_env_beat_defaults() {
let d = parse(&argv(""), &no_env).unwrap();
assert_eq!(d.falkor, "redis://127.0.0.1:6379");
assert_eq!(d.graph, "graph");
assert_eq!(d.listen, "127.0.0.1:50056");
assert!(d.tables.is_empty() && d.basic.is_none() && d.tls.is_none() && d.lifecycle.is_none());
let env = |v: &str| match v {
"GRAPHAR_FLIGHT_ADDR" => Some("0.0.0.0:1".to_string()),
"GRAPHAR_FLIGHT_GRAPH" => Some("from_env".to_string()),
_ => None,
};
let c = parse(
&argv("--listen 0.0.0.0:50056 --falkor redis://falkor:6379 --token T --basic u:p --table people=MATCH_(n:Person)_RETURN_n._gar_id_AS__gar_id,_n.name_AS_name --tls-cert c.pem --tls-key k.pem --system"),
&env,
)
.unwrap();
assert_eq!(c.listen, "0.0.0.0:50056", "the flag beats the env");
assert_eq!(c.graph, "from_env", "the env beats the default");
assert_eq!(c.falkor, "redis://falkor:6379");
assert_eq!(c.token_flag.as_deref(), Some("T"));
assert_eq!(c.basic, Some(("u".into(), "p".into())));
assert_eq!(c.tables.len(), 1);
assert_eq!(c.tables[0].0, "people");
assert_eq!(c.tls, Some(("c.pem".into(), "k.pem".into(), None)));
assert!(c.system);
}
/// **The lifecycle verbs are recognised, and a half TLS pair or a bad
/// `--table` is refused rather than defaulted.**
#[test]
fn lifecycle_verbs_are_seen_and_malformed_flags_are_refused() {
for v in LIFECYCLE {
assert_eq!(parse(&argv(v), &no_env).unwrap().lifecycle.as_deref(), Some(v));
}
assert!(parse(&argv("--tls-cert c.pem"), &no_env).is_err(), "cert without key");
assert!(parse(&argv("--table people"), &no_env).is_err(), "no `=`");
assert!(parse(&argv("--basic nocolon"), &no_env).is_err());
}
/// **A `--table` becomes an advertised table with the columns its RETURN
/// names, typed by the id rule; no `--table` advertises the two generic ones.**
/// RED direction: make `return_columns` ignore `AS` and the alias assertion
/// fails; register nothing for the empty case and the count is 0.
#[test]
fn tables_advertise_their_return_columns_and_the_default_pair_exists() {
assert_eq!(
return_columns("MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name AS name, n.age ORDER BY n.age LIMIT 5"),
vec!["_gar_id", "name", "n.age"]
);
assert!(return_columns("MATCH (n) DELETE n").is_empty());
let r = build_registry(&[(
"people".into(),
"MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name AS name".into(),
)]);
let tables = r.tables();
assert_eq!(tables.len(), 1);
let (name, schema) = &tables[0];
assert_eq!(name, "people");
assert_eq!(schema.field(0).name(), "_gar_id");
assert_eq!(schema.field(0).data_type(), &DataType::Int64);
assert_eq!(schema.field(1).name(), "name");
assert_eq!(schema.field(1).data_type(), &DataType::Utf8);
let generic = build_registry(&[]);
let mut names: Vec<String> = generic.tables().into_iter().map(|(n, _)| n).collect();
names.sort();
assert_eq!(names, vec!["graph_edges", "graph_nodes"]);
}
}