#![allow(clippy::result_large_err)]
use std::env;
use reqwest::Client;
use serde_json::{Value as Json, json};
use crate::backend::BackendKind;
use crate::runtime::executor_utils::build_probe;
use crate::runtime::executors::{
BackendExecutor, BackendHealth, BackendProbe, MutationExecutor, ObjectExecutor, QueryExecutor,
ResourceAdminExecutor, SearchExecutor,
};
fn validate_neo4j_identifier(id: &str) -> Result<(), String> {
if id.is_empty() || id.len() > 64 {
return Err(format!(
"Neo4j identifier '{id}' is invalid: must be 1–64 characters"
));
}
let first = id.chars().next().unwrap();
if !first.is_ascii_alphabetic() && first != '_' {
return Err(format!(
"Neo4j identifier '{id}' must start with a letter or underscore"
));
}
if !id.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
return Err(format!(
"Neo4j identifier '{id}' contains invalid characters; \
only ASCII letters, digits, and underscores are allowed"
));
}
Ok(())
}
#[derive(Debug, Clone)]
pub struct Neo4jConfig {
pub http_base: String,
pub username: String,
pub password: String,
pub database: String,
pub is_cloud: bool,
pub dev_mode: bool,
pub timeout_secs: u64,
}
impl Neo4jConfig {
pub fn from_env() -> Option<Self> {
let dsn = env::var("UDB_GRAPH_DSN").ok();
let http_override = env::var("UDB_GRAPH_HTTP_URL").ok();
let http_base = if let Some(url) = http_override {
url
} else if let Some(ref dsn) = dsn {
Self::http_base_from_dsn(dsn)
} else {
return None;
};
let username = env::var("UDB_GRAPH_USER").unwrap_or_else(|_| "neo4j".to_string());
let password = env::var("UDB_GRAPH_PASSWORD").unwrap_or_default();
let database = env::var("UDB_GRAPH_DATABASE").unwrap_or_else(|_| "neo4j".to_string());
let is_cloud =
super::http::is_cloud("UDB_NEO4J_DEPLOY_MODE", &http_base, ".databases.neo4j.io");
let dev_mode = std::env::var("UDB_DEV_MODE")
.map(|v| matches!(v.as_str(), "1" | "true" | "yes"))
.unwrap_or(false);
let timeout_secs = super::http::env_timeout("UDB_GRAPH_TIMEOUT_SECS", 30).as_secs();
Some(Self {
http_base,
username,
password,
database,
is_cloud,
dev_mode,
timeout_secs,
})
}
pub(crate) fn http_base_from_dsn(dsn: &str) -> String {
if dsn.starts_with("http://") || dsn.starts_with("https://") {
if let Some(at) = dsn.find('@') {
let scheme_end = dsn.find("://").map(|i| i + 3).unwrap_or(0);
let scheme = &dsn[..scheme_end];
return format!("{scheme}{}", &dsn[at + 1..]);
}
return dsn.to_string();
}
let rest = dsn.strip_prefix("bolt://").unwrap_or(dsn);
let rest = if rest.contains('@') {
rest.split_once('@').map(|(_, r)| r).unwrap_or(rest)
} else {
rest
};
let host_port = rest.split('/').next().unwrap_or(rest);
let (host, _) = host_port.split_once(':').unwrap_or((host_port, "7687"));
format!("http://{host}:7474")
}
}
#[derive(Debug, Clone)]
pub struct Neo4jExecutor {
config: Neo4jConfig,
http: Client,
}
impl crate::runtime::backend_context::BackendContextEnforcer for Neo4jExecutor {
fn backend_label(&self) -> &str {
"neo4j"
}
fn enforce(
&self,
ctx: &crate::runtime::backend_context::AppliedContext,
) -> crate::runtime::backend_context::ContextEffect {
crate::runtime::backend_context::enforce_with_mechanism(
ctx,
"_tenant_id / _project_id stamped in MERGE key; ANDed into MATCH WHERE",
)
}
}
impl Neo4jExecutor {
pub fn new(config: Neo4jConfig) -> Self {
if config.is_cloud && config.http_base.starts_with("http://") {
tracing::error!(
http_base = %config.http_base,
"Neo4j is configured as cloud but the HTTP base uses http:// — \
all requests will fail. Change to https:// or set UDB_NEO4J_DEPLOY_MODE=self_hosted"
);
}
if !config.dev_mode && !config.is_cloud && config.http_base.starts_with("http://") {
tracing::warn!(
http_base = %config.http_base,
"Neo4j HTTP base uses plain HTTP — Basic Auth credentials will be sent unencrypted"
);
}
let timeout = std::time::Duration::from_secs(config.timeout_secs.max(1));
let http = super::http::HttpClientSpec::with_timeout(timeout)
.https_only(config.is_cloud)
.build();
Self { config, http }
}
pub fn kind(&self) -> BackendKind {
BackendKind::Neo4j
}
pub fn name(&self) -> &str {
"Neo4j"
}
pub fn from_env() -> Option<Self> {
Neo4jConfig::from_env().map(Self::new)
}
fn tx_url(&self) -> String {
format!(
"{}/db/{}/tx/commit",
self.config.http_base, self.config.database
)
}
pub async fn cypher(&self, statements: &[(&str, Json)]) -> Result<Vec<Json>, String> {
let stmt_array: Vec<Json> = statements
.iter()
.map(|(text, params)| json!({ "statement": text, "parameters": params }))
.collect();
let body = json!({ "statements": stmt_array });
let resp = self
.http
.post(self.tx_url())
.basic_auth(&self.config.username, Some(&self.config.password))
.header("Content-Type", "application/json")
.header("Accept", "application/json;charset=UTF-8")
.json(&body)
.send()
.await
.map_err(|e| format!("Neo4j HTTP error: {e}"))?;
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
let parsed: Json = serde_json::from_str(&text)
.map_err(|e| format!("Neo4j response decode failed: {e}"))?;
if let Some(errors) = parsed.get("errors").and_then(|e| e.as_array())
&& !errors.is_empty()
{
let msg = errors
.iter()
.filter_map(|e| e.get("message").and_then(|m| m.as_str()))
.collect::<Vec<_>>()
.join("; ");
return Err(format!("Neo4j Cypher error: {msg}"));
}
if !status.is_success() {
return Err(format!("Neo4j HTTP [{status}]: {text}"));
}
let results = parsed
.get("results")
.and_then(|r| r.as_array())
.cloned()
.unwrap_or_default();
Ok(results)
}
async fn run_single(&self, cypher: &str, params: Json) -> Result<Vec<Json>, String> {
let results = self.cypher(&[(cypher, params)]).await?;
let rows = results
.into_iter()
.next()
.and_then(|r| {
let columns = r
.get("columns")
.and_then(|c| c.as_array())
.cloned()
.unwrap_or_default();
let data = r.get("data").and_then(|d| d.as_array()).cloned()?;
Some(
data.into_iter()
.map(|row| {
let row_vals = row
.get("row")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
let mut obj = serde_json::Map::new();
for (col, val) in columns.iter().zip(row_vals.iter()) {
if let Some(name) = col.as_str() {
obj.insert(name.to_string(), val.clone());
}
}
Json::Object(obj)
})
.collect::<Vec<_>>(),
)
})
.unwrap_or_default();
Ok(rows)
}
pub async fn create_node(&self, label: &str, id: &str, properties: Json) -> Result<(), String> {
validate_neo4j_identifier(label)?;
let cypher = format!("MERGE (n:{label} {{id: $id}}) SET n += $props RETURN n",);
self.run_single(&cypher, json!({ "id": id, "props": properties }))
.await?;
Ok(())
}
pub async fn find_nodes(
&self,
label: &str,
filter: Json,
limit: i64,
) -> Result<Vec<Json>, String> {
validate_neo4j_identifier(label)?;
let limit_clause = if limit > 0 {
format!(" LIMIT {limit}")
} else {
String::new()
};
let cypher = format!(
"MATCH (n:{label}) WHERE all(k IN keys($f) WHERE n[k] = $f[k]) RETURN n{limit_clause}"
);
self.run_single(&cypher, json!({ "f": filter })).await
}
pub async fn update_node(&self, label: &str, id: &str, properties: Json) -> Result<(), String> {
validate_neo4j_identifier(label)?;
let cypher = format!("MATCH (n:{label} {{id: $id}}) SET n += $props RETURN n");
self.run_single(&cypher, json!({ "id": id, "props": properties }))
.await?;
Ok(())
}
pub async fn delete_node(&self, label: &str, id: &str) -> Result<(), String> {
validate_neo4j_identifier(label)?;
let cypher = format!("MATCH (n:{label} {{id: $id}}) DETACH DELETE n");
self.run_single(&cypher, json!({ "id": id })).await?;
Ok(())
}
pub async fn create_relationship(
&self,
from_label: &str,
from_id: &str,
to_label: &str,
to_id: &str,
rel_type: &str,
properties: Json,
) -> Result<(), String> {
validate_neo4j_identifier(from_label)?;
validate_neo4j_identifier(to_label)?;
validate_neo4j_identifier(rel_type)?;
let cypher = format!(
"MATCH (a:{from_label} {{id: $from_id}}), (b:{to_label} {{id: $to_id}}) \
MERGE (a)-[r:{rel_type}]->(b) SET r += $props RETURN r"
);
self.run_single(
&cypher,
json!({ "from_id": from_id, "to_id": to_id, "props": properties }),
)
.await?;
Ok(())
}
}
impl BackendHealth for Neo4jExecutor {
async fn ping(&self) -> Result<(), String> {
self.run_single("RETURN 1 AS ok", json!({})).await?;
Ok(())
}
}
impl QueryExecutor for Neo4jExecutor {
async fn query(&self, request_json: &str) -> Result<String, tonic::Status> {
let spec: Json = serde_json::from_str(request_json)
.map_err(|e| tonic::Status::invalid_argument(format!("invalid request json: {e}")))?;
let rows = if let Some(cypher) = spec.get("cypher").and_then(Json::as_str) {
let params = spec.get("parameters").cloned().unwrap_or_else(|| json!({}));
self.cypher(&[(cypher, params)])
.await
.map_err(tonic::Status::internal)?
} else {
let label = spec
.get("label")
.and_then(Json::as_str)
.ok_or_else(|| tonic::Status::invalid_argument("missing required field 'label'"))?;
let filter = spec.get("filter").cloned().unwrap_or_else(|| json!({}));
let limit = spec.get("limit").and_then(Json::as_i64).unwrap_or(100);
self.find_nodes(label, filter, limit)
.await
.map_err(tonic::Status::internal)?
};
serde_json::to_string(&rows).map_err(|e| tonic::Status::internal(e.to_string()))
}
}
impl MutationExecutor for Neo4jExecutor {
async fn mutate(&self, request_json: &str) -> Result<String, tonic::Status> {
let spec: Json = serde_json::from_str(request_json)
.map_err(|e| tonic::Status::invalid_argument(format!("invalid request json: {e}")))?;
let operation = spec
.get("operation")
.and_then(Json::as_str)
.ok_or_else(|| tonic::Status::invalid_argument("missing required field 'operation'"))?;
let req_str = |key: &str| -> Result<String, tonic::Status> {
spec.get(key)
.and_then(Json::as_str)
.map(|s| s.to_string())
.ok_or_else(|| {
tonic::Status::invalid_argument(format!("missing required field '{key}'"))
})
};
match operation {
"cypher" => {
let cypher = req_str("cypher")?;
let params = spec.get("parameters").cloned().unwrap_or_else(|| json!({}));
let rows = self
.cypher(&[(&cypher, params)])
.await
.map_err(tonic::Status::internal)?;
serde_json::to_string(&rows).map_err(|e| tonic::Status::internal(e.to_string()))
}
"create_node" | "upsert_node" => {
let label = req_str("label")?;
let id = req_str("id")?;
let properties = spec
.get("properties")
.or_else(|| spec.get("props"))
.cloned()
.unwrap_or_else(|| json!({}));
self.create_node(&label, &id, properties)
.await
.map_err(tonic::Status::internal)?;
Ok(r#"{"affected_rows":1}"#.to_string())
}
"update_node" => {
let label = req_str("label")?;
let id = req_str("id")?;
let properties = spec.get("properties").cloned().unwrap_or_else(|| json!({}));
self.update_node(&label, &id, properties)
.await
.map_err(tonic::Status::internal)?;
Ok(r#"{"affected_rows":1}"#.to_string())
}
"delete_node" => {
let label = req_str("label")?;
let id = req_str("id")?;
self.delete_node(&label, &id)
.await
.map_err(tonic::Status::internal)?;
Ok(r#"{"affected_rows":1}"#.to_string())
}
"create_relationship" | "upsert_relationship" => {
let from_label = req_str("from_label")?;
let from_id = req_str("from_id")?;
let to_label = req_str("to_label")?;
let to_id = req_str("to_id")?;
let rel_type = req_str("rel_type")?;
let properties = spec.get("properties").cloned().unwrap_or_else(|| json!({}));
self.create_relationship(
&from_label,
&from_id,
&to_label,
&to_id,
&rel_type,
properties,
)
.await
.map_err(tonic::Status::internal)?;
Ok(r#"{"affected_rows":1}"#.to_string())
}
other => Err(tonic::Status::invalid_argument(format!(
"unsupported Neo4j mutation operation '{other}'"
))),
}
}
}
impl SearchExecutor for Neo4jExecutor {
async fn search(&self, _request_json: &str) -> Result<String, tonic::Status> {
Err(tonic::Status::failed_precondition(
"neo4j does not support generic vector search dispatch",
))
}
}
impl ObjectExecutor for Neo4jExecutor {
async fn get_object(&self, _request_json: &str) -> Result<Vec<u8>, tonic::Status> {
Err(tonic::Status::failed_precondition(
"neo4j is not an object store",
))
}
async fn put_object(
&self,
_request_json: &str,
_bytes: Vec<u8>,
) -> Result<String, tonic::Status> {
Err(tonic::Status::failed_precondition(
"neo4j is not an object store",
))
}
}
impl ResourceAdminExecutor for Neo4jExecutor {
async fn ensure_resource(
&self,
resource_name: &str,
spec_json: &str,
) -> Result<(), tonic::Status> {
validate_neo4j_identifier(resource_name).map_err(tonic::Status::internal)?;
let spec: Json = serde_json::from_str(spec_json).unwrap_or(json!({}));
let prop = spec
.get("constraint_property")
.and_then(|v| v.as_str())
.unwrap_or("id");
validate_neo4j_identifier(prop).map_err(tonic::Status::internal)?;
let constraint_name = format!("udb_{resource_name}_{prop}_unique");
let cypher = format!(
"CREATE CONSTRAINT {constraint_name} IF NOT EXISTS \
FOR (n:{resource_name}) REQUIRE n.{prop} IS UNIQUE"
);
self.run_single(&cypher, json!({}))
.await
.map(|_| ())
.map_err(tonic::Status::internal)
}
async fn drop_resource(&self, resource_name: &str) -> Result<(), tonic::Status> {
validate_neo4j_identifier(resource_name).map_err(tonic::Status::internal)?;
let cypher = format!("DROP CONSTRAINT udb_{resource_name}_id_unique IF EXISTS");
self.run_single(&cypher, json!({}))
.await
.map(|_| ())
.map_err(tonic::Status::internal)
}
async fn list_resources(&self) -> Result<Vec<String>, tonic::Status> {
let rows = self
.run_single("SHOW CONSTRAINTS WHERE name STARTS WITH 'udb_'", json!({}))
.await
.map_err(tonic::Status::internal)?;
let names = rows
.iter()
.filter_map(|r| r.get("name").and_then(|v| v.as_str()))
.map(|s| s.to_string())
.collect();
Ok(names)
}
}
impl BackendExecutor for Neo4jExecutor {
async fn transaction(&self, _request_json: &str) -> Result<String, tonic::Status> {
Err(tonic::Status::failed_precondition(
"neo4j transactions are not exposed via generic dispatch",
))
}
async fn probe(&self) -> Result<BackendProbe, tonic::Status> {
Ok(build_probe(
"neo4j",
<Self as BackendHealth>::ping(self).await,
))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn neo4j_executor_kind_and_name() {
let cfg = Neo4jConfig {
http_base: "http://localhost:7474".to_string(),
username: "neo4j".to_string(),
password: "secret".to_string(),
database: "neo4j".to_string(),
is_cloud: false,
dev_mode: true,
timeout_secs: 30,
};
let exec = Neo4jExecutor::new(cfg);
assert_eq!(exec.kind(), BackendKind::Neo4j);
assert_eq!(exec.name(), "Neo4j");
}
#[tokio::test]
async fn neo4j_backend_executor_rejects_unsupported_and_malformed() {
let exec = Neo4jExecutor::new(Neo4jConfig {
http_base: "http://localhost:7474".to_string(),
username: "neo4j".to_string(),
password: "secret".to_string(),
database: "neo4j".to_string(),
is_cloud: false,
dev_mode: true,
timeout_secs: 30,
});
assert!(SearchExecutor::search(&exec, "{}").await.is_err());
assert!(ObjectExecutor::get_object(&exec, "{}").await.is_err());
assert!(BackendExecutor::transaction(&exec, "{}").await.is_err());
assert!(QueryExecutor::query(&exec, "not json").await.is_err());
assert!(MutationExecutor::mutate(&exec, "{}").await.is_err());
assert!(
MutationExecutor::mutate(&exec, r#"{"operation":"bogus"}"#)
.await
.is_err()
);
}
#[test]
fn neo4j_tx_url_correct() {
let cfg = Neo4jConfig {
http_base: "http://localhost:7474".to_string(),
username: "neo4j".to_string(),
password: "".to_string(),
database: "neo4j".to_string(),
is_cloud: false,
dev_mode: true,
timeout_secs: 30,
};
let exec = Neo4jExecutor::new(cfg);
assert_eq!(exec.tx_url(), "http://localhost:7474/db/neo4j/tx/commit");
}
#[test]
fn neo4j_config_bolt_to_http() {
let http = Neo4jConfig::http_base_from_dsn("bolt://user:pass@graph.example.com:7687");
assert_eq!(http, "http://graph.example.com:7474");
}
#[test]
fn neo4j_config_http_passthrough() {
let http = Neo4jConfig::http_base_from_dsn("http://auradb.neo4j.io:7474");
assert_eq!(http, "http://auradb.neo4j.io:7474");
}
#[test]
fn neo4j_config_from_env_returns_none_without_vars() {
unsafe {
env::remove_var("UDB_GRAPH_DSN");
env::remove_var("UDB_GRAPH_HTTP_URL");
}
assert!(Neo4jConfig::from_env().is_none());
}
}