crates/graphar-flight/src/bin/graphar-graph-http.rs
//! `graphar-graph-http` — a few NAMED read queries over one FalkorDB graph, as Arrow IPC over
//! plain HTTP, for a browser: a wasm viewer cannot speak Flight's gRPC, and must not speak
//! Cypher to the database at all.
//!
//! **Not a query service.** The client never sends query text. Each route is one Cypher
//! statement written here; the only inputs are label/type names (identifier characters only),
//! integer ids and an integer limit, checked before they are spliced in. Every statement runs as
//! `GRAPH.RO_QUERY`, so the server refuses a write whatever happens here, and the ACL user it
//! connects as needs `+graph.ro_query` on the one graph and nothing else.
//!
//! **Not public.** `--listen` is a private address and `--allow` the callers that may connect
//! (the front door's address); anyone else is answered 403 before a query is run. The door in
//! front of it is what checks the person (same as the power endpoint).
//!
//! ```text
//! graphar-graph-http --listen 172.16.12.19:8095 --allow 172.16.12.4 \
//! --url rediss://172.16.12.19:6379 --graph simulated \
//! --tls-cert viewer.crt --tls-key viewer.key --tls-ca ca-chain.crt
//! ```
//!
//! Routes (also under any `--prefix`, default `/sim/api`):
//! * `GET meta.json` — `{"round", "persons", "nodes", "edges"}`
//! * `GET nodes.arrow?labels=A,B&ids=1,2&limit=N` — `id, label, name, kommun, round`
//! * `GET edges.arrow?types=T,U&limit=N` — `src, dst, type`
//! * `GET ego.arrow?id=N` — the edges between a node and its neighbours: `src, dst, type`
use std::net::IpAddr;
use std::sync::Arc;
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow_array::RecordBatch;
use graphar_flight::falkor::FalkorExecutor;
const DEFAULT_LIMIT: usize = 200_000;
const MAX_LIMIT: usize = 1_000_000;
const EGO_LIMIT: usize = 5_000;
fn main() {
let args: Vec<String> = std::env::args().collect();
let opt = |k: &str| args.windows(2).find(|w| w[0] == k).map(|w| w[1].clone());
if args.iter().any(|a| a == "--help" || a == "-h") {
println!("{}", HELP.trim());
return;
}
let need = |k: &str| opt(k).unwrap_or_else(|| usage(&format!("{k} is required")));
let listen = need("--listen");
let allow: Vec<IpAddr> = need("--allow")
.split(',')
.map(|a| a.trim().parse().unwrap_or_else(|_| usage(&format!("--allow: {a:?} is not an address"))))
.collect();
if allow.is_empty() {
usage("--allow needs at least one caller");
}
let prefix = opt("--prefix").unwrap_or_else(|| "/sim/api".into());
let url = opt("--url").unwrap_or_else(|| "redis://127.0.0.1:6379".into());
let graph = need("--graph");
let tls = match (opt("--tls-cert"), opt("--tls-key")) {
(Some(c), Some(k)) => Some(
graphar_falkordb::FalkorTls::from_files(c.as_ref(), k.as_ref(), opt("--tls-ca").as_deref().map(std::path::Path::new))
.unwrap_or_else(|e| usage(&e.to_string())),
),
(None, None) => None,
_ => usage("--tls-cert and --tls-key go together"),
};
let rt = Arc::new(tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("runtime"));
let exec = rt
.block_on(FalkorExecutor::connect_with_identity(&url, graph.clone(), tls))
.unwrap_or_else(|e| {
eprintln!("graphar-graph-http: cannot connect to {url}: {e}");
std::process::exit(1);
});
let server = tiny_http::Server::http(&listen).unwrap_or_else(|e| {
eprintln!("graphar-graph-http: cannot listen on {listen}: {e}");
std::process::exit(1);
});
eprintln!("graphar-graph-http: {listen}, graph `{graph}` at {url}, callers {allow:?}, prefix {prefix}");
// One thread per core, all on the same blocking listener: a viewer asks a handful of
// things at once (nodes, edges, meta), and none of them waits for another.
let server = Arc::new(server);
let workers: Vec<_> = (0..std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4))
.map(|_| {
let (server, rt, exec, allow, prefix) = (server.clone(), rt.clone(), exec.clone(), allow.clone(), prefix.clone());
std::thread::spawn(move || {
for req in server.incoming_requests() {
let mut exec = exec.clone();
let resp = handle(&req, &allow, &prefix, |c, s| rt.block_on(exec.ro_query(c, s)));
let _ = req.respond(resp);
}
})
})
.collect();
for w in workers {
let _ = w.join();
}
}
type Resp = tiny_http::Response<std::io::Cursor<Vec<u8>>>;
fn handle(
req: &tiny_http::Request,
allow: &[IpAddr],
prefix: &str,
mut query: impl FnMut(&str, &SchemaRef) -> graphar_flight::error::Result<RecordBatch>,
) -> Resp {
let caller = req.remote_addr().map(|a| a.ip());
let (path, q) = req.url().split_once('?').unwrap_or((req.url(), ""));
let route = path.strip_prefix(prefix).unwrap_or(path).trim_start_matches('/').to_string();
let out = (|| -> Result<Resp, (u16, String)> {
if !caller.is_some_and(|c| allow.contains(&c)) {
return Err((403, format!("caller {caller:?} is not allowed")));
}
if *req.method() != tiny_http::Method::Get {
return Err((405, "GET only".into()));
}
let params = Params::parse(q)?;
let mut run = |cypher: &str, schema: &SchemaRef| query(cypher, schema).map_err(|e| (502, format!("falkordb: {e}")));
match route.as_str() {
"meta.json" => {
let one = Arc::new(Schema::new(vec![Field::new("n", DataType::Int64, true)]));
let scalar = |b: RecordBatch| {
use arrow_array::Array;
let c = b.column(0).as_any().downcast_ref::<arrow_array::Int64Array>().cloned();
c.and_then(|c| (c.len() > 0 && !c.is_null(0)).then(|| c.value(0))).unwrap_or(0)
};
let round = scalar(run("MATCH (n) WHERE n.round IS NOT NULL RETURN max(n.round)", &one)?);
let persons = scalar(run("MATCH (n:Person) RETURN count(n)", &one)?);
let nodes = scalar(run("MATCH (n) RETURN count(n)", &one)?);
let edges = scalar(run("MATCH ()-[e]->() RETURN count(e)", &one)?);
let body = format!(r#"{{"round":{round},"persons":{persons},"nodes":{nodes},"edges":{edges}}}"#);
Ok(bytes(200, "application/json", body.into_bytes()))
}
"nodes.arrow" => {
let limit = params.limit()?;
let ret = "RETURN ID(n), labels(n)[0], n.name, n.kommun, coalesce(n.round, 0)";
let cypher = if let Some(ids) = params.ints("ids")? {
format!("MATCH (n) WHERE ID(n) IN [{}] {ret} LIMIT {limit}", join(&ids))
} else {
match params.names("labels")?.as_deref() {
None => format!("MATCH (n:Person) {ret} LIMIT {limit}"),
Some([one]) => format!("MATCH (n:`{one}`) {ret} LIMIT {limit}"),
Some(many) => format!("MATCH (n) WHERE labels(n)[0] IN [{}] {ret} LIMIT {limit}", quoted(many)),
}
};
arrow(run(&cypher, &node_schema())?)
}
"edges.arrow" => {
let limit = params.limit()?;
// A typed pattern, not `WHERE type(e) IN [..]`: the filter scans every edge
// (6 s for one type of 1.1 M), the pattern reads that type's matrix alone.
let rel = match params.names("types")? {
Some(t) => format!(":{}", t.iter().map(|n| format!("`{n}`")).collect::<Vec<_>>().join("|")),
None => String::new(),
};
arrow(run(&format!("MATCH (a)-[e{rel}]->(b) RETURN ID(a), ID(b), type(e) LIMIT {limit}"), &edge_schema())?)
}
"ego.arrow" => {
let id = params.ints("id")?.and_then(|v| v.first().copied()).ok_or((400, "ego.arrow needs ?id=<integer>".to_string()))?;
let out = run(&format!("MATCH (a)-[e]->(b) WHERE ID(a) = {id} RETURN ID(a), ID(b), type(e) LIMIT {EGO_LIMIT}"), &edge_schema())?;
let inn = run(&format!("MATCH (a)-[e]->(b) WHERE ID(b) = {id} RETURN ID(a), ID(b), type(e) LIMIT {EGO_LIMIT}"), &edge_schema())?;
let both = arrow::compute::concat_batches(&edge_schema(), [&out, &inn]).map_err(|e| (500, e.to_string()))?;
arrow(both)
}
_ => Err((404, format!("no route {path}"))),
}
})();
let resp = out.unwrap_or_else(|(code, msg)| bytes(code, "text/plain; charset=utf-8", msg.into_bytes()));
eprintln!("{} {} {} -> {}", caller.map(|c| c.to_string()).unwrap_or_default(), req.method(), req.url(), resp.status_code().0);
resp
}
fn node_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("label", DataType::Utf8, true),
Field::new("name", DataType::Utf8, true),
Field::new("kommun", DataType::Utf8, true),
Field::new("round", DataType::Int64, true),
]))
}
fn edge_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("type", DataType::Utf8, true),
]))
}
fn arrow(batch: RecordBatch) -> Result<Resp, (u16, String)> {
let mut buf = Vec::with_capacity(batch.get_array_memory_size());
{
let mut w = arrow_ipc::writer::StreamWriter::try_new(&mut buf, &batch.schema()).map_err(|e| (500, e.to_string()))?;
w.write(&batch).map_err(|e| (500, e.to_string()))?;
w.finish().map_err(|e| (500, e.to_string()))?;
}
Ok(bytes(200, "application/vnd.apache.arrow.stream", buf))
}
fn bytes(code: u16, ctype: &str, body: Vec<u8>) -> Resp {
tiny_http::Response::from_data(body)
.with_status_code(code)
.with_header(tiny_http::Header::from_bytes("Content-Type", ctype).unwrap())
.with_header(tiny_http::Header::from_bytes("Cache-Control", "no-store").unwrap())
}
/// The query string, every value checked before it can reach a statement.
struct Params(Vec<(String, String)>);
impl Params {
fn parse(q: &str) -> Result<Self, (u16, String)> {
Ok(Params(
q.split('&')
.filter(|p| !p.is_empty())
.map(|p| {
let (k, v) = p.split_once('=').unwrap_or((p, ""));
(k.to_string(), v.replace("%2C", ",").replace("%2c", ","))
})
.collect(),
))
}
fn get(&self, k: &str) -> Option<&str> {
self.0.iter().find(|(key, _)| key == k).map(|(_, v)| v.as_str()).filter(|v| !v.is_empty())
}
/// Identifier names only - `[A-Za-z_][A-Za-z0-9_]*` - so a name can be spliced into a
/// statement and cannot end it.
fn names(&self, k: &str) -> Result<Option<Vec<String>>, (u16, String)> {
let Some(v) = self.get(k) else { return Ok(None) };
let names: Vec<String> = v.split(',').map(str::to_string).collect();
for n in &names {
let ok = n.chars().next().is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
&& n.chars().all(|c| c.is_ascii_alphanumeric() || c == '_');
if !ok {
return Err((400, format!("{k}: {n:?} is not a name")));
}
}
Ok(Some(names))
}
fn ints(&self, k: &str) -> Result<Option<Vec<i64>>, (u16, String)> {
let Some(v) = self.get(k) else { return Ok(None) };
v.split(',').map(|x| x.parse::<i64>().map_err(|_| (400, format!("{k}: {x:?} is not an integer")))).collect::<Result<Vec<_>, _>>().map(Some)
}
fn limit(&self) -> Result<usize, (u16, String)> {
match self.get("limit") {
None => Ok(DEFAULT_LIMIT),
Some(v) => v.parse::<usize>().map(|n| n.min(MAX_LIMIT)).map_err(|_| (400, format!("limit: {v:?} is not an integer"))),
}
}
}
fn join(ids: &[i64]) -> String {
ids.iter().map(i64::to_string).collect::<Vec<_>>().join(",")
}
fn quoted(names: &[String]) -> String {
names.iter().map(|n| format!("'{n}'")).collect::<Vec<_>>().join(",")
}
fn usage(why: &str) -> ! {
eprintln!("graphar-graph-http: {why}\n\n{}", HELP.trim());
std::process::exit(2);
}
const HELP: &str = "
graphar-graph-http - named read queries over one FalkorDB graph, as Arrow IPC over HTTP
--listen ADDR private address:port to serve on (required)
--allow IP,IP the callers allowed to connect - the front door (required)
--graph NAME the graph (required)
--url URL redis:// or rediss:// (default redis://127.0.0.1:6379)
--tls-cert PEM --tls-key PEM [--tls-ca PEM] mutual TLS; CN = a read-only ACL user
--prefix PATH route prefix (default /sim/api)
GET meta.json | nodes.arrow?labels=&ids=&limit= | edges.arrow?types=&limit= | ego.arrow?id=
";