use aetherdb::{Key, QueryResult, Value};
use thiserror::Error;
use super::{Aetherdb, JtiStore, State};
const MAX_KEY_BYTES: usize = 4 * 1024;
const MAX_VALUE_BYTES: usize = 64 * 1024;
const MAX_QUERY_TEXT: usize = 16 * 1024;
const MAX_RESULT_ROWS: usize = 64;
#[derive(Debug, Error)]
pub enum BusDbError {
#[error("AetherDB bus API unavailable (KWT_REPLAY=0 or open failed)")]
Disabled,
#[error("key exceeds {MAX_KEY_BYTES} bytes")]
KeyTooLarge,
#[error("value exceeds {MAX_VALUE_BYTES} bytes")]
ValueTooLarge,
#[error("query text exceeds {MAX_QUERY_TEXT} bytes")]
QueryTooLarge,
#[error("mutations require NAUTALID_AETHERDB_BUS_MUTATE=1")]
MutateDisabled,
#[error("missing required KDL property `{0}`")]
MissingProperty(String),
#[error("AetherDB: {0}")]
Db(#[from] aetherdb::Error),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BusDbStatus {
pub enabled: bool,
pub shared: bool,
pub mode: String,
pub shards: u32,
pub replication: u32,
pub node_id: u64,
pub jti_prefix: String,
pub blades: Vec<&'static str>,
}
impl JtiStore {
pub fn bus_status(&self) -> Result<BusDbStatus, BusDbError> {
self.bus_status_with_topology(&super::topology::Topology::from_env())
}
pub fn bus_status_with_topology(
&self,
topology: &super::topology::Topology,
) -> Result<BusDbStatus, BusDbError> {
self.with_db(|db| {
let blades = db.db.capabilities().blades().names();
let preset_shards = db.db.preset().shards;
Ok(BusDbStatus {
enabled: true,
shared: db.shared,
mode: topology.mode_label().into(),
shards: preset_shards.max(topology.shards),
replication: db.db.preset().replication.factor,
node_id: topology.node_id,
jti_prefix: db.prefix.clone(),
blades,
})
})
}
pub fn bus_get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, BusDbError> {
validate_key(key)?;
self.with_db(|db| {
let key = Key::from_bytes(key.to_vec());
match db.db.get(&key)? {
Some(value) => Ok(Some(value.as_bytes().to_vec())),
None => Ok(None),
}
})
}
pub fn bus_put(&self, key: &[u8], value: &[u8]) -> Result<(), BusDbError> {
ensure_mutate_enabled()?;
validate_key(key)?;
validate_value(value)?;
self.with_db(|db| {
let key = Key::from_bytes(key.to_vec());
let value = Value::from_bytes(value.to_vec());
db.db.put(key, value)?;
Ok(())
})
}
pub fn bus_delete(&self, key: &[u8]) -> Result<(), BusDbError> {
ensure_mutate_enabled()?;
validate_key(key)?;
self.with_db(|db| {
let key = Key::from_bytes(key.to_vec());
db.db.delete(&key)?;
Ok(())
})
}
pub fn bus_query(&self, text: &str) -> Result<String, BusDbError> {
validate_query(text)?;
ensure_query_mutate_allowed(text)?;
self.with_db(|db| {
let result = db.db.queries().execute(text)?;
Ok(format_query_result(&result))
})
}
pub fn bus_sql(&self, text: &str) -> Result<String, BusDbError> {
validate_query(text)?;
ensure_query_mutate_allowed(text)?;
self.with_db(|db| {
let result = db.db.queries().sql(text)?;
Ok(format_query_result(&result))
})
}
pub fn bus_cypher(&self, text: &str) -> Result<String, BusDbError> {
validate_query(text)?;
ensure_query_mutate_allowed(text)?;
self.with_db(|db| {
let result = db.db.queries().cypher(text)?;
Ok(format_query_result(&result))
})
}
fn with_db<R>(&self, f: impl FnOnce(&Aetherdb) -> Result<R, BusDbError>) -> Result<R, BusDbError> {
let guard = self.state.lock().expect("jti store lock poisoned");
match &*guard {
State::Disabled => Err(BusDbError::Disabled),
State::Db(db) => f(db),
}
}
}
fn ensure_mutate_enabled() -> Result<(), BusDbError> {
if crate::surfaces::env_enabled("NAUTALID_AETHERDB_BUS_MUTATE") {
Ok(())
} else {
Err(BusDbError::MutateDisabled)
}
}
fn validate_key(key: &[u8]) -> Result<(), BusDbError> {
if key.len() > MAX_KEY_BYTES {
return Err(BusDbError::KeyTooLarge);
}
Ok(())
}
fn validate_value(value: &[u8]) -> Result<(), BusDbError> {
if value.len() > MAX_VALUE_BYTES {
return Err(BusDbError::ValueTooLarge);
}
Ok(())
}
fn validate_query(text: &str) -> Result<(), BusDbError> {
if text.len() > MAX_QUERY_TEXT {
return Err(BusDbError::QueryTooLarge);
}
Ok(())
}
fn ensure_query_mutate_allowed(text: &str) -> Result<(), BusDbError> {
if query_appears_read_only(text) {
Ok(())
} else {
ensure_mutate_enabled()
}
}
fn query_appears_read_only(text: &str) -> bool {
let upper = text.trim().to_ascii_uppercase();
if upper.is_empty() {
return true;
}
if ["SELECT", "SHOW", "DESCRIBE", "DESC", "EXPLAIN", "PRAGMA"]
.iter()
.any(|prefix| upper.starts_with(prefix))
{
return true;
}
upper.starts_with("MATCH")
&& !upper.contains("CREATE")
&& !upper.contains("DELETE")
&& !upper.contains("SET")
&& !upper.contains("MERGE")
&& !upper.contains("REMOVE")
}
pub fn format_bytes_attr(label: &str, bytes: &[u8]) -> String {
if bytes.is_empty() {
return format!("{label}=\"\"");
}
if bytes
.iter()
.all(|b| b.is_ascii_graphic() || *b == b' ' || *b == b'\t')
{
let escaped = String::from_utf8_lossy(bytes)
.replace('\\', "\\\\")
.replace('"', "\\\"");
format!("{label}=\"{escaped}\"")
} else {
format!("{label}_hex=\"{}\"", hex::encode(bytes))
}
}
pub fn format_query_result_public(result: &QueryResult) -> String {
format_query_result(result)
}
fn format_query_result(result: &QueryResult) -> String {
match result {
QueryResult::Rows(rows) => {
let mut out = format!("rows count={}", rows.len());
for (index, (key, value)) in rows.iter().take(MAX_RESULT_ROWS).enumerate() {
out.push(' ');
out.push_str(&format_bytes_attr(&format!("key{index}"), key.as_bytes()));
out.push(' ');
out.push_str(&format_bytes_attr(&format!("value{index}"), value.as_bytes()));
}
if rows.len() > MAX_RESULT_ROWS {
out.push_str(" truncated=true");
}
out
}
QueryResult::Nodes(ids) => format!("nodes count={}", ids.len()),
QueryResult::Inserted { id } => format!("ok inserted id={id}"),
QueryResult::Affected(count) => format!("ok affected={count}"),
QueryResult::Scalar(value) => format!("ok scalar={value}"),
QueryResult::Empty => "ok empty".into(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use aetherdb::open_bundled;
use std::sync::Mutex;
use super::{Aetherdb, JtiStore, State};
static MUTATE_ENV_LOCK: Mutex<()> = Mutex::new(());
fn kv_store() -> JtiStore {
JtiStore {
state: std::sync::Arc::new(std::sync::Mutex::new(State::Db(Aetherdb {
db: open_bundled("kv-local").expect("kv-local"),
prefix: "test:bus:".into(),
ttl_secs: 3600,
shared: false,
max_jtis: 1_000_000,
}))),
}
}
#[test]
fn bus_get_put_roundtrip() {
let _guard = MUTATE_ENV_LOCK.lock().expect("mutate env lock");
unsafe {
std::env::set_var("NAUTALID_AETHERDB_BUS_MUTATE", "1");
}
let store = kv_store();
store.bus_put(b"bus:hello", b"world").expect("put");
let value = store.bus_get(b"bus:hello").expect("get").expect("some");
assert_eq!(value, b"world");
store.bus_delete(b"bus:hello").expect("delete");
assert!(store.bus_get(b"bus:hello").expect("get").is_none());
unsafe {
std::env::remove_var("NAUTALID_AETHERDB_BUS_MUTATE");
}
}
#[test]
fn mutate_requires_env() {
let _guard = MUTATE_ENV_LOCK.lock().expect("mutate env lock");
unsafe {
std::env::remove_var("NAUTALID_AETHERDB_BUS_MUTATE");
}
let store = kv_store();
assert!(matches!(
store.bus_put(b"k", b"v"),
Err(BusDbError::MutateDisabled)
));
}
#[test]
fn mutating_sql_requires_mutate_env() {
let _guard = MUTATE_ENV_LOCK.lock().expect("mutate env lock");
unsafe {
std::env::remove_var("NAUTALID_AETHERDB_BUS_MUTATE");
}
let store = kv_store();
assert!(matches!(
store.bus_sql("DELETE FROM kv WHERE 1=1"),
Err(BusDbError::MutateDisabled)
));
}
#[test]
fn select_sql_allowed_without_mutate_env() {
let _guard = MUTATE_ENV_LOCK.lock().expect("mutate env lock");
unsafe {
std::env::remove_var("NAUTALID_AETHERDB_BUS_MUTATE");
}
let store = kv_store();
let result = store.bus_sql("SELECT 1");
assert!(!matches!(result, Err(BusDbError::MutateDisabled)));
}
}