facett-simview/src/contract.rs
//! **The HTTP contract** the page speaks — fixed with the server side (the bench's
//! door + benchpower for the presses, a small graph endpoint for the data). One
//! definition, read by the app, by the mock and by the dev server, so the three
//! cannot drift.
//!
//! | request | method · path | answer |
//! |---|---|---|
//! | Generate | `POST /power/falkordb/simulate` | 202 accepted · 409 a job is running |
//! | Permutate 10% | `POST /power/falkordb/permutate` | as above |
//! | job state | `GET /power` | JSON `{"benches":[{"id":"falkordb","busy":..,"last":{..}}]}` |
//! | meta | `GET /sim/api/meta.json` | JSON `{"round","persons","nodes","edges"}` |
//! | nodes | `GET /sim/api/nodes.arrow?labels=Person&limit=N` / `?ids=1,2` | Arrow IPC stream: `id` i64, `label`, `name`, `kommun`? utf8, `round` i64 |
//! | edges | `GET /sim/api/edges.arrow?types=..&limit=N` | Arrow IPC stream: `src`, `dst` i64, `type` utf8 |
//! | ego | `GET /sim/api/ego.arrow?id=N` | the edges of the node and its 1-hop neighbours |
use std::io::Cursor;
use std::sync::Arc;
use arrow_array::{Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
/// The bench target every press is for.
pub const TARGET: &str = "falkordb";
/// The two press verbs.
pub const VERB_GENERATE: &str = "simulate";
pub const VERB_PERMUTATE: &str = "permutate";
/// How many Person nodes the 3D view asks for.
pub const NODE_LIMIT: usize = 200_000;
/// What the app asks for.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Req {
Press(&'static str),
Power,
Meta,
Persons,
Ego(i64),
NodesByIds(Vec<i64>),
}
impl Req {
/// `(method, url)` - absolute paths, same origin.
pub fn http(&self) -> (&'static str, String) {
match self {
Req::Press(verb) => ("POST", format!("/power/{TARGET}/{verb}")),
Req::Power => ("GET", "/power".into()),
Req::Meta => ("GET", "/sim/api/meta.json".into()),
Req::Persons => ("GET", format!("/sim/api/nodes.arrow?labels=Person&limit={NODE_LIMIT}")),
Req::Ego(id) => ("GET", format!("/sim/api/ego.arrow?id={id}")),
Req::NodesByIds(ids) => (
"GET",
format!("/sim/api/nodes.arrow?ids={}", ids.iter().map(i64::to_string).collect::<Vec<_>>().join(",")),
),
}
}
}
/// What came back: the request, the HTTP status, the body.
#[derive(Debug, Clone)]
pub struct Resp {
pub req: Req,
pub status: u16,
pub body: Vec<u8>,
}
/// `GET /sim/api/meta.json`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Meta {
pub round: i64,
pub persons: i64,
pub nodes: i64,
pub edges: i64,
}
impl Meta {
pub fn parse(body: &[u8]) -> Option<Meta> {
let v: serde_json::Value = serde_json::from_slice(body).ok()?;
let n = |k: &str| v.get(k).and_then(serde_json::Value::as_i64).unwrap_or(0);
Some(Meta { round: n("round"), persons: n("persons"), nodes: n("nodes"), edges: n("edges") })
}
pub fn to_json(self) -> String {
serde_json::json!({"round": self.round, "persons": self.persons, "nodes": self.nodes, "edges": self.edges}).to_string()
}
}
/// The `falkordb` entry of `GET /power`.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Job {
pub busy: bool,
pub verb: Option<String>,
pub for_s: Option<i64>,
/// `(verb, ok, detail)` of the last finished job.
pub last: Option<(String, bool, String)>,
}
impl Job {
pub fn parse(body: &[u8]) -> Option<Job> {
let v: serde_json::Value = serde_json::from_slice(body).ok()?;
let b = v.get("benches")?.as_array()?.iter().find(|b| b.get("id").and_then(|x| x.as_str()) == Some(TARGET))?;
let s = |x: &serde_json::Value, k: &str| x.get(k).and_then(|y| y.as_str()).map(str::to_string);
Some(Job {
busy: b.get("busy").and_then(serde_json::Value::as_bool).unwrap_or(false),
verb: s(b, "verb"),
for_s: b.get("for_s").and_then(serde_json::Value::as_i64),
last: b.get("last").filter(|l| !l.is_null()).map(|l| {
(
s(l, "verb").unwrap_or_default(),
l.get("ok").and_then(serde_json::Value::as_bool).unwrap_or(false),
s(l, "detail").unwrap_or_default(),
)
}),
})
}
}
/// One node row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NodeRow {
pub id: i64,
pub label: String,
pub name: String,
pub kommun: Option<String>,
pub round: i64,
}
/// One edge row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EdgeRow {
pub src: i64,
pub dst: i64,
pub etype: String,
}
pub fn node_schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("label", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("kommun", DataType::Utf8, true),
Field::new("round", DataType::Int64, false),
]))
}
pub fn edge_schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("type", DataType::Utf8, false),
]))
}
fn batches(body: &[u8]) -> Result<Vec<RecordBatch>, String> {
let r = arrow_ipc::reader::StreamReader::try_new(Cursor::new(body), None).map_err(|e| format!("arrow stream: {e}"))?;
r.collect::<Result<Vec<_>, _>>().map_err(|e| format!("arrow stream: {e}"))
}
fn col<'a, T: 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T, String> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<T>())
.ok_or_else(|| format!("column `{name}` missing or of the wrong type"))
}
/// Decode a `nodes.arrow` answer.
pub fn decode_nodes(body: &[u8]) -> Result<Vec<NodeRow>, String> {
let mut out = Vec::new();
for b in batches(body)? {
let (id, label, name, round) =
(col::<Int64Array>(&b, "id")?, col::<StringArray>(&b, "label")?, col::<StringArray>(&b, "name")?, col::<Int64Array>(&b, "round")?);
let kommun = b.column_by_name("kommun").and_then(|c| c.as_any().downcast_ref::<StringArray>());
for i in 0..b.num_rows() {
out.push(NodeRow {
id: id.value(i),
label: label.value(i).to_string(),
name: name.value(i).to_string(),
kommun: kommun.filter(|k| !k.is_null(i)).map(|k| k.value(i).to_string()),
round: round.value(i),
});
}
}
Ok(out)
}
/// Decode an `edges.arrow` / `ego.arrow` answer.
pub fn decode_edges(body: &[u8]) -> Result<Vec<EdgeRow>, String> {
let mut out = Vec::new();
for b in batches(body)? {
let (src, dst, t) = (col::<Int64Array>(&b, "src")?, col::<Int64Array>(&b, "dst")?, col::<StringArray>(&b, "type")?);
for i in 0..b.num_rows() {
out.push(EdgeRow { src: src.value(i), dst: dst.value(i), etype: t.value(i).to_string() });
}
}
Ok(out)
}
fn stream(schema: Arc<Schema>, batch: RecordBatch) -> Vec<u8> {
let mut buf = Vec::new();
{
let mut w = arrow_ipc::writer::StreamWriter::try_new(&mut buf, &schema).expect("stream writer");
w.write(&batch).expect("write batch");
w.finish().expect("finish stream");
}
buf
}
/// Encode nodes as the server answers them (the mock and the dev server use this).
pub fn encode_nodes(rows: &[NodeRow]) -> Vec<u8> {
let s = node_schema();
let batch = RecordBatch::try_new(
s.clone(),
vec![
Arc::new(Int64Array::from(rows.iter().map(|r| r.id).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.label.as_str()).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.name.as_str()).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.kommun.as_deref()).collect::<Vec<_>>())),
Arc::new(Int64Array::from(rows.iter().map(|r| r.round).collect::<Vec<_>>())),
],
)
.expect("node batch");
stream(s, batch)
}
/// Encode edges as the server answers them.
pub fn encode_edges(rows: &[EdgeRow]) -> Vec<u8> {
let s = edge_schema();
let batch = RecordBatch::try_new(
s.clone(),
vec![
Arc::new(Int64Array::from(rows.iter().map(|r| r.src).collect::<Vec<_>>())),
Arc::new(Int64Array::from(rows.iter().map(|r| r.dst).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.etype.as_str()).collect::<Vec<_>>())),
],
)
.expect("edge batch");
stream(s, batch)
}