use super::packstream::{Value, MAX_PACKSTREAM_DEPTH};
use serde_json::Value as Json;
pub const TAG_NODE: u8 = 0x4E;
pub const TAG_RELATIONSHIP: u8 = 0x52;
pub const TAG_UNBOUND_RELATIONSHIP: u8 = 0x72;
pub const TAG_PATH: u8 = 0x50;
pub fn json_to_value(json: &Json) -> Result<Value, String> {
to_value(json, 1)
}
fn to_value(json: &Json, depth: usize) -> Result<Value, String> {
Ok(match json {
Json::Null => Value::Null,
Json::Bool(b) => Value::Bool(*b),
Json::Number(n) => {
if let Some(i) = n.as_i64() {
Value::Int(i)
} else if n.is_u64() {
return Err(format!(
"integer {n} does not fit a PackStream INT_64 (max {})",
i64::MAX
));
} else {
Value::Float(n.as_f64().unwrap_or(f64::NAN))
}
}
Json::String(s) => Value::String(s.clone()),
Json::Array(items) => {
check_depth(depth)?;
Value::List(
items
.iter()
.map(|v| to_value(v, depth + 1))
.collect::<Result<_, _>>()?,
)
}
Json::Object(map) => {
check_depth(depth)?;
if map.len() == 1 {
if let Some(node) = map.get("$node") {
return node_value(node, depth + 1).map(|n| n.value);
}
if let Some(rel) = map.get("$relationship") {
return relationship_value(rel, depth + 1).map(|r| r.bound);
}
if let Some(path) = map.get("$path") {
return path_value(path, depth + 1);
}
}
Value::Map(
map.iter()
.map(|(k, v)| Ok((k.clone(), to_value(v, depth + 1)?)))
.collect::<Result<_, String>>()?,
)
}
})
}
fn check_depth(depth: usize) -> Result<(), String> {
if depth > MAX_PACKSTREAM_DEPTH {
Err(format!(
"value nests deeper than {MAX_PACKSTREAM_DEPTH} levels"
))
} else {
Ok(())
}
}
fn object<'a>(json: &'a Json, what: &str) -> Result<&'a serde_json::Map<String, Json>, String> {
json.as_object()
.ok_or_else(|| format!("{what} must be an object"))
}
fn int_field(map: &serde_json::Map<String, Json>, key: &str, what: &str) -> Result<i64, String> {
map.get(key)
.and_then(Json::as_i64)
.ok_or_else(|| format!("{what} needs an integer '{key}'"))
}
fn element_id(map: &serde_json::Map<String, Json>, key: &str, fallback: i64) -> Value {
Value::String(
map.get(key)
.and_then(Json::as_str)
.map(str::to_string)
.unwrap_or_else(|| fallback.to_string()),
)
}
fn properties(
map: &serde_json::Map<String, Json>,
what: &str,
depth: usize,
) -> Result<Value, String> {
match map.get("properties") {
None | Some(Json::Null) => Ok(Value::Map(Vec::new())),
Some(Json::Object(_)) => to_value(&map["properties"], depth),
Some(_) => Err(format!("{what} 'properties' must be an object")),
}
}
struct Node {
id: i64,
value: Value,
}
fn node_value(json: &Json, depth: usize) -> Result<Node, String> {
check_depth(depth)?;
let json = json.get("$node").unwrap_or(json);
let map = object(json, "$node")?;
let id = int_field(map, "id", "$node")?;
let labels = match map.get("labels") {
None | Some(Json::Null) => Vec::new(),
Some(Json::Array(labels)) => labels
.iter()
.map(|l| {
l.as_str()
.map(Value::string)
.ok_or_else(|| "$node 'labels' must be strings".to_string())
})
.collect::<Result<_, _>>()?,
Some(_) => return Err("$node 'labels' must be an array of strings".to_string()),
};
Ok(Node {
id,
value: Value::Struct {
tag: TAG_NODE,
fields: vec![
Value::Int(id),
Value::List(labels),
properties(map, "$node", depth + 1)?,
element_id(map, "element_id", id),
],
},
})
}
struct Relationship {
id: i64,
start: i64,
end: i64,
bound: Value,
unbound: Value,
}
fn relationship_value(json: &Json, depth: usize) -> Result<Relationship, String> {
check_depth(depth)?;
let json = json.get("$relationship").unwrap_or(json);
let map = object(json, "$relationship")?;
let id = int_field(map, "id", "$relationship")?;
let start = int_field(map, "start", "$relationship")?;
let end = int_field(map, "end", "$relationship")?;
let rel_type = map
.get("type")
.and_then(Json::as_str)
.filter(|t| !t.is_empty())
.ok_or("$relationship needs a non-empty string 'type'")?
.to_string();
let props = properties(map, "$relationship", depth + 1)?;
let eid = element_id(map, "element_id", id);
Ok(Relationship {
id,
start,
end,
bound: Value::Struct {
tag: TAG_RELATIONSHIP,
fields: vec![
Value::Int(id),
Value::Int(start),
Value::Int(end),
Value::String(rel_type.clone()),
props.clone(),
eid.clone(),
element_id(map, "start_element_id", start),
element_id(map, "end_element_id", end),
],
},
unbound: Value::Struct {
tag: TAG_UNBOUND_RELATIONSHIP,
fields: vec![Value::Int(id), Value::String(rel_type), props, eid],
},
})
}
fn path_value(json: &Json, depth: usize) -> Result<Value, String> {
check_depth(depth)?;
let map = object(json, "$path")?;
let nodes = map
.get("nodes")
.and_then(Json::as_array)
.ok_or("$path needs a 'nodes' array")?;
let rels = match map.get("relationships") {
None | Some(Json::Null) => Vec::new(),
Some(Json::Array(r)) => r.clone(),
Some(_) => return Err("$path 'relationships' must be an array".to_string()),
};
if nodes.is_empty() {
return Err("$path needs at least one node".to_string());
}
if rels.len() + 1 != nodes.len() {
return Err(format!(
"$path with {} nodes needs {} relationships, got {}",
nodes.len(),
nodes.len() - 1,
rels.len()
));
}
let nodes: Vec<Node> = nodes
.iter()
.map(|n| node_value(n, depth + 1))
.collect::<Result<_, _>>()?;
let rels: Vec<Relationship> = rels
.iter()
.map(|r| relationship_value(r, depth + 1))
.collect::<Result<_, _>>()?;
let mut unique_nodes: Vec<&Node> = Vec::new();
for node in &nodes {
if !unique_nodes.iter().any(|n| n.id == node.id) {
unique_nodes.push(node);
}
}
let mut unique_rels: Vec<&Relationship> = Vec::new();
for rel in &rels {
if !unique_rels.iter().any(|r| r.id == rel.id) {
unique_rels.push(rel);
}
}
let mut indices = Vec::with_capacity(rels.len() * 2);
for (i, rel) in rels.iter().enumerate() {
let (from, to) = (nodes[i].id, nodes[i + 1].id);
let rel_index = unique_rels
.iter()
.position(|r| r.id == rel.id)
.map(|p| p as i64 + 1)
.unwrap_or(1);
let signed = if rel.start == from && rel.end == to {
rel_index
} else if rel.start == to && rel.end == from {
-rel_index
} else {
return Err(format!(
"$path relationship {} ({}->{}) does not connect nodes {} and {}",
rel.id, rel.start, rel.end, from, to
));
};
let node_index = unique_nodes
.iter()
.position(|n| n.id == to)
.map(|p| p as i64)
.unwrap_or(0);
indices.push(Value::Int(signed));
indices.push(Value::Int(node_index));
}
Ok(Value::Struct {
tag: TAG_PATH,
fields: vec![
Value::List(unique_nodes.iter().map(|n| n.value.clone()).collect()),
Value::List(unique_rels.iter().map(|r| r.unbound.clone()).collect()),
Value::List(indices),
],
})
}
pub fn structure_name(tag: u8) -> String {
match tag {
TAG_NODE => "Node".to_string(),
TAG_RELATIONSHIP => "Relationship".to_string(),
TAG_UNBOUND_RELATIONSHIP => "UnboundRelationship".to_string(),
TAG_PATH => "Path".to_string(),
0x44 => "Date".to_string(),
0x54 => "Time".to_string(),
0x74 => "LocalTime".to_string(),
0x49 => "DateTime".to_string(),
0x69 => "DateTimeZoneId".to_string(),
0x64 => "LocalDateTime".to_string(),
0x45 => "Duration".to_string(),
0x58 => "Point2D".to_string(),
0x59 => "Point3D".to_string(),
other => format!("0x{other:02X}"),
}
}
pub fn value_to_json(value: &Value) -> Json {
match value {
Value::Null => Json::Null,
Value::Bool(b) => Json::Bool(*b),
Value::Int(i) => Json::from(*i),
Value::Float(f) => serde_json::Number::from_f64(*f)
.map(Json::Number)
.unwrap_or_else(|| Json::String(f.to_string())),
Value::Bytes(b) => serde_json::json!({"$bytes_length": b.len()}),
Value::String(s) => Json::String(s.clone()),
Value::List(items) => Json::Array(items.iter().map(value_to_json).collect()),
Value::Map(entries) => Json::Object(
entries
.iter()
.map(|(k, v)| (k.clone(), value_to_json(v)))
.collect(),
),
Value::Struct { tag, fields } => serde_json::json!({
"$structure": structure_name(*tag),
"fields": fields.iter().map(value_to_json).collect::<Vec<_>>(),
}),
}
}
pub const STAT_COUNTERS: &[&str] = &[
"nodes-created",
"nodes-deleted",
"relationships-created",
"relationships-deleted",
"properties-set",
"labels-added",
"labels-removed",
"indexes-added",
"indexes-removed",
"constraints-added",
"constraints-removed",
"system-updates",
];
pub fn stats_value(json: Option<&Json>) -> Result<Option<Value>, String> {
let map = match json {
None | Some(Json::Null) => return Ok(None),
Some(Json::Object(map)) => map,
Some(_) => return Err("'stats' must be an object".to_string()),
};
let mut entries = Vec::new();
let mut updates = false;
let mut system_updates = false;
for (key, value) in map {
let wire = key.replace('_', "-");
if !STAT_COUNTERS.contains(&wire.as_str()) {
return Err(format!(
"unknown stats key '{key}' (allowed: {})",
STAT_COUNTERS
.iter()
.map(|k| k.replace('-', "_"))
.collect::<Vec<_>>()
.join(", ")
));
}
let count = value
.as_i64()
.filter(|c| *c >= 0)
.ok_or_else(|| format!("stats '{key}' must be a non-negative integer"))?;
if count == 0 {
continue;
}
if wire == "system-updates" {
system_updates = true;
} else {
updates = true;
}
let wire_key = STAT_COUNTERS
.iter()
.find(|k| **k == wire)
.copied()
.unwrap_or("nodes-created");
entries.push((wire_key.to_string(), Value::Int(count)));
}
if entries.is_empty() {
return Ok(None);
}
if updates {
entries.push(("contains-updates".to_string(), Value::Bool(true)));
}
if system_updates {
entries.push(("contains-system-updates".to_string(), Value::Bool(true)));
}
Ok(Some(Value::Map(entries)))
}
pub fn valid_failure_code(code: &str) -> bool {
let parts: Vec<&str> = code.split('.').collect();
let word = |s: &str| {
!s.is_empty()
&& s.len() <= 64
&& s.chars().next().is_some_and(|c| c.is_ascii_uppercase())
&& s.chars().all(|c| c.is_ascii_alphanumeric())
};
parts.len() == 4
&& parts[0] == "Neo"
&& matches!(parts[1], "ClientError" | "TransientError" | "DatabaseError")
&& word(parts[2])
&& word(parts[3])
}
#[derive(Debug, Clone)]
pub struct QueryAnswer {
pub fields: Vec<String>,
pub records: Vec<Vec<Value>>,
pub stats: Option<Value>,
pub query_type: &'static str,
}
pub const MAX_ACTION_DEPTH: usize = MAX_PACKSTREAM_DEPTH + 4;
pub const MAX_ACTION_NODES: usize = 65536;
pub const MAX_ACTION_RETAINED_BYTES: usize = 8 * 1024 * 1024;
pub fn action_within_budget(action: &Json) -> bool {
crate::utils::json_budget::within_budget(
action,
MAX_ACTION_RETAINED_BYTES,
MAX_ACTION_NODES,
MAX_ACTION_DEPTH,
)
}
fn within_wire_depth(value: &Value, depth: usize) -> bool {
let mut pending = vec![(value, depth)];
while let Some((value, depth)) = pending.pop() {
let children = match value {
Value::List(v) | Value::Struct { fields: v, .. } => Some(v.as_slice()),
Value::Map(v) => {
if depth > MAX_PACKSTREAM_DEPTH {
return false;
}
pending.extend(v.iter().map(|(_, v)| (v, depth + 1)));
None
}
_ => None,
};
if let Some(children) = children {
if depth > MAX_PACKSTREAM_DEPTH {
return false;
}
pending.extend(children.iter().map(|v| (v, depth + 1)));
}
}
true
}
pub fn query_answer(action: &Json) -> Result<QueryAnswer, String> {
if !action_within_budget(action) {
return Err("Bolt answer exceeds the in-memory depth/node/retained-content budget".into());
}
let fields: Vec<String> = match action.get("fields") {
Some(Json::Array(fields)) => fields
.iter()
.map(|f| {
f.as_str()
.filter(|s| !s.is_empty())
.map(str::to_string)
.ok_or_else(|| "every entry in 'fields' must be a non-empty string".to_string())
})
.collect::<Result<_, _>>()?,
None | Some(Json::Null) => Vec::new(),
Some(_) => return Err("'fields' must be an array of column names".to_string()),
};
for (i, f) in fields.iter().enumerate() {
if fields[..i].contains(f) {
return Err(format!("column '{f}' appears twice in 'fields'"));
}
}
let rows = match action.get("records") {
None | Some(Json::Null) => Vec::new(),
Some(Json::Array(rows)) => rows.iter().collect::<Vec<_>>(),
Some(_) => return Err("'records' must be an array of rows".to_string()),
};
let mut records = Vec::with_capacity(rows.len());
for (i, row) in rows.iter().enumerate() {
let cells = row.as_array().ok_or_else(|| {
format!("record {i} must be an array with one value per field, in field order")
})?;
if cells.len() != fields.len() {
return Err(format!(
"record {i} has {} values but there are {} fields",
cells.len(),
fields.len()
));
}
records.push(
cells
.iter()
.map(|c| {
let value = json_to_value(c).map_err(|e| format!("record {i}: {e}"))?;
if !within_wire_depth(&value, 3) {
return Err(format!(
"record {i}: encoded value exceeds wire depth {MAX_PACKSTREAM_DEPTH}"
));
}
Ok(value)
})
.collect::<Result<Vec<_>, _>>()?,
);
}
let query_type = match action.get("query_type").and_then(Json::as_str) {
None => "r",
Some("r") => "r",
Some("w") => "w",
Some("rw") => "rw",
Some("s") => "s",
Some(other) => {
return Err(format!(
"query_type must be r (read), w (write), rw (read-write) or s (schema), got \
'{other}'"
))
}
};
Ok(QueryAnswer {
fields,
records,
stats: stats_value(action.get("stats"))?,
query_type,
})
}