use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::{Arc, Mutex};
use cljrs_tx::{HostApi, TxError, TxLimits, TxProgram};
use cljrs_value::clone::SerializedValue;
use corium_core::{EntityId, IndexOrder, Keyword, TotalF64, Value as CoreValue};
use corium_db::Db;
use corium_query::QInput;
use corium_query::boundary::{edn_to_value, value_to_edn};
use corium_query::edn::Edn;
use thiserror::Error;
use crate::node::TxFnExpander;
#[derive(Clone, Copy, Debug)]
pub struct DbFnBudget {
pub fuel: u64,
pub memory_bytes: usize,
pub call_depth: u64,
}
impl Default for DbFnBudget {
fn default() -> Self {
let defaults = TxLimits::default();
Self {
fuel: defaults.gas,
memory_bytes: defaults.memory_bytes,
call_depth: defaults.call_depth,
}
}
}
pub const INVOCATION_STACK_BYTES: usize = 64 * 1024 * 1024;
#[derive(Debug, Error)]
pub enum DbFnError {
#[error("db function failed: {0}")]
Execution(#[from] TxError),
#[error("db function recursion limit ({0}) exceeded")]
Recursion(usize),
#[error("db function must return a sequence of tx forms, got {0}")]
BadResult(String),
#[error("db function boundary conversion failed: {0}")]
Convert(String),
}
const NATIVE_OPS: &[&str] = &["db/add", "db/retract", "db/cas", "db/retractEntity"];
pub struct DbFnExpander {
budget: DbFnBudget,
max_depth: usize,
programs: Mutex<HashMap<u64, Arc<TxProgram>>>,
}
impl Default for DbFnExpander {
fn default() -> Self {
Self::new(DbFnBudget::default())
}
}
impl DbFnExpander {
#[must_use]
pub fn new(budget: DbFnBudget) -> Self {
Self {
budget,
max_depth: 16,
programs: Mutex::new(HashMap::new()),
}
}
#[must_use]
pub fn with_max_depth(mut self, max_depth: usize) -> Self {
self.max_depth = max_depth;
self
}
pub fn expand(&self, db: &Db, forms: Vec<Edn>) -> Result<Vec<Edn>, DbFnError> {
let host = read_api(db);
self.expand_at(db, &host, forms, self.max_depth)
}
fn expand_at(
&self,
db: &Db,
host: &HostApi,
forms: Vec<Edn>,
depth: usize,
) -> Result<Vec<Edn>, DbFnError> {
let mut out = Vec::with_capacity(forms.len());
for form in forms {
match invocation(db, &form) {
None => out.push(form),
Some((source, args)) => {
if depth == 0 {
return Err(DbFnError::Recursion(self.max_depth));
}
let produced = self.invoke(db, host, &source, &args)?;
out.extend(self.expand_at(db, host, produced, depth - 1)?);
}
}
}
Ok(out)
}
fn invoke(
&self,
db: &Db,
host: &HostApi,
source: &str,
args: &[Edn],
) -> Result<Vec<Edn>, DbFnError> {
let program = self.program(source)?;
let mut call_args = Vec::with_capacity(args.len() + 1);
call_args.push(db_token(db));
for arg in args {
call_args.push(edn_to_sv(arg));
}
let limits = TxLimits {
memory_bytes: self.budget.memory_bytes,
gas: self.budget.fuel,
call_depth: self.budget.call_depth,
};
let result = std::thread::scope(|scope| {
std::thread::Builder::new()
.name("corium-txfn".into())
.stack_size(INVOCATION_STACK_BYTES)
.spawn_scoped(scope, || {
cljrs_tx::execute_with_host(&program, call_args, limits, host)
})
.map_err(|error| {
DbFnError::Convert(format!("cannot spawn invocation thread: {error}"))
})?
.join()
.map_err(|_| DbFnError::Convert("invocation thread panicked".into()))?
.map_err(DbFnError::from)
})?;
let items = match &result {
SerializedValue::List(items) | SerializedValue::Vector(items) => items.clone(),
SerializedValue::Nil => Vec::new(),
other => return Err(DbFnError::BadResult(format!("{other:?}"))),
};
items
.iter()
.map(|item| sv_to_edn(item).map_err(DbFnError::Convert))
.collect()
}
fn program(&self, source: &str) -> Result<Arc<TxProgram>, DbFnError> {
let mut hasher = DefaultHasher::new();
source.hash(&mut hasher);
let key = hasher.finish();
let mut programs = self
.programs
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(program) = programs.get(&key) {
return Ok(Arc::clone(program));
}
let program = Arc::new(TxProgram::parse(source)?);
programs.insert(key, Arc::clone(&program));
Ok(program)
}
}
impl TxFnExpander for DbFnExpander {
fn expand(&self, db: &Db, forms: Vec<Edn>) -> Result<Vec<Edn>, String> {
Self::expand(self, db, forms).map_err(|error| error.to_string())
}
}
fn invocation(db: &Db, form: &Edn) -> Option<(String, Vec<Edn>)> {
let (Edn::Vector(items) | Edn::List(items)) = form else {
return None;
};
let head = items.first()?.as_keyword()?;
let full = head.namespace.as_deref().map_or_else(
|| head.name.clone(),
|namespace| format!("{namespace}/{}", head.name),
);
if NATIVE_OPS.contains(&full.as_str()) {
return None;
}
let source = fn_source(db, head)?;
Some((source, items[1..].to_vec()))
}
fn fn_source(db: &Db, ident: &Keyword) -> Option<String> {
let fn_attr = db.idents().entid(&Keyword::parse("db/fn"))?;
let entity = db.idents().entid(ident).or_else(|| {
let ident_attr = db.idents().entid(&Keyword::parse("db/ident"))?;
let interned = db.interner().get(ident)?;
db.lookup(ident_attr, &CoreValue::Keyword(interned))
})?;
match db.values(entity, fn_attr).into_iter().next() {
Some(CoreValue::Str(source)) => Some(source.to_string()),
_ => None,
}
}
const TOKEN_BASIS: &str = "basis-t";
const TOKEN_AS_OF: &str = "as-of";
const TOKEN_SINCE: &str = "since";
const TOKEN_HISTORY: &str = "history";
fn token_key(name: &str) -> SerializedValue {
SerializedValue::Keyword {
namespace: Some("corium.db".into()),
name: name.into(),
}
}
fn is_keyword(value: &SerializedValue, namespace: Option<&str>, name: &str) -> bool {
matches!(
value,
SerializedValue::Keyword {
namespace: kw_namespace,
name: kw_name,
} if kw_namespace.as_deref() == namespace && &**kw_name == name
)
}
#[derive(Clone, Copy, Default)]
struct TokenView {
as_of: Option<u64>,
since: Option<u64>,
history: bool,
}
impl TokenView {
fn apply(self, db: &Db) -> Db {
let mut view = db.clone();
if let Some(t) = self.as_of {
view = view.as_of(t);
}
if let Some(t) = self.since {
view = view.since(t);
}
if self.history {
view = view.history();
}
view
}
fn token(self, db: &Db) -> SerializedValue {
let optional_t = |t: Option<u64>| {
t.map_or(SerializedValue::Nil, |t| {
SerializedValue::Long(i64::try_from(t).unwrap_or(i64::MAX))
})
};
SerializedValue::ArrayMap(vec![
(
token_key(TOKEN_BASIS),
SerializedValue::Long(i64::try_from(db.basis_t()).unwrap_or(i64::MAX)),
),
(token_key(TOKEN_AS_OF), optional_t(self.as_of)),
(token_key(TOKEN_SINCE), optional_t(self.since)),
(
token_key(TOKEN_HISTORY),
SerializedValue::Bool(self.history),
),
])
}
}
fn db_token(db: &Db) -> SerializedValue {
TokenView::default().token(db)
}
fn token_view(value: &SerializedValue) -> Option<TokenView> {
let (SerializedValue::ArrayMap(pairs)
| SerializedValue::HashMap(pairs)
| SerializedValue::SortedMap(pairs)) = value
else {
return None;
};
let field = |name: &str| {
pairs
.iter()
.find(|(key, _)| is_keyword(key, Some("corium.db"), name))
.map(|(_, value)| value)
};
field(TOKEN_BASIS)?;
let t_of = |name: &str| match field(name) {
Some(SerializedValue::Long(t)) => u64::try_from(*t).ok(),
_ => None,
};
Some(TokenView {
as_of: t_of(TOKEN_AS_OF),
since: t_of(TOKEN_SINCE),
history: matches!(field(TOKEN_HISTORY), Some(SerializedValue::Bool(true))),
})
}
#[allow(clippy::too_many_lines)]
fn read_api(db: &Db) -> HostApi {
let mut host = HostApi::new();
let arity = |args: &[SerializedValue], want: usize, what: &str| {
if args.len() == want {
Ok(())
} else {
Err(format!("{what} takes {want} arguments, got {}", args.len()))
}
};
let view_arg = |db: &Db, value: &SerializedValue, what: &str| {
token_view(value)
.map(|view| view.apply(db))
.ok_or_else(|| format!("{what} takes a database as its first argument"))
};
let base = db.clone();
host.define("corium.api", "q", move |args: &[SerializedValue]| {
if args.len() < 2 {
return Err("q takes a query and at least one input".into());
}
let query = sv_to_edn(&args[0])?;
let inputs: Vec<Result<Db, Edn>> = args[1..]
.iter()
.map(|arg| match token_view(arg) {
Some(view) => Ok(Ok(view.apply(&base))),
None => sv_to_edn(arg).map(Err),
})
.collect::<Result<_, _>>()?;
let qinputs: Vec<QInput<'_>> = inputs
.iter()
.map(|input| match input {
Ok(db) => QInput::Db(db),
Err(edn) => QInput::Edn(edn.clone()),
})
.collect();
let result = corium_query::q(&query, &qinputs).map_err(|error| error.to_string())?;
Ok(edn_to_sv(&result))
});
let base = db.clone();
host.define("corium.api", "pull", move |args: &[SerializedValue]| {
arity(args, 3, "pull")?;
let db = view_arg(&base, &args[0], "pull")?;
let pattern = sv_to_edn(&args[1])?;
let eid = eid_of(&db, &sv_to_edn(&args[2])?)?;
let result = corium_query::pull(&db, &pattern, eid).map_err(|error| error.to_string())?;
Ok(edn_to_sv(&result))
});
let base = db.clone();
host.define("corium.api", "entity", move |args: &[SerializedValue]| {
arity(args, 2, "entity")?;
let db = view_arg(&base, &args[0], "entity")?;
let eid = eid_of(&db, &sv_to_edn(&args[1])?)?;
let entity = corium_query::Entity::new(&db, eid);
let mut pairs = vec![(
Edn::keyword("db/id"),
Edn::Long(i64::try_from(eid.raw()).unwrap_or(i64::MAX)),
)];
for attr in entity.keys() {
let Some(keyword) = db.idents().ident(attr) else {
continue;
};
let values = entity.get(attr);
let many = db
.schema()
.get(attr)
.is_some_and(|meta| meta.cardinality == corium_core::Cardinality::Many);
let value = if many {
let mut items: Vec<Edn> = values
.iter()
.map(|value| value_to_edn(&db, value))
.collect();
items.sort();
items.dedup();
Edn::Set(items)
} else if let Some(value) = values.first() {
value_to_edn(&db, value)
} else {
continue;
};
pairs.push((Edn::Keyword(keyword.clone()), value));
}
pairs.sort_by(|a, b| a.0.cmp(&b.0));
Ok(edn_to_sv(&Edn::Map(pairs)))
});
let base = db.clone();
host.define("corium.api", "datoms", move |args: &[SerializedValue]| {
arity(args, 2, "datoms")?;
let db = view_arg(&base, &args[0], "datoms")?;
let index = sv_to_edn(&args[1])?;
let order = match index.as_keyword().map(|keyword| keyword.name.as_str()) {
Some("eavt") => IndexOrder::Eavt,
Some("aevt") => IndexOrder::Aevt,
Some("avet") => IndexOrder::Avet,
Some("vaet") => IndexOrder::Vaet,
_ => return Err("index must be :eavt, :aevt, :avet, or :vaet".into()),
};
let items: Vec<Edn> = db
.datoms_at(order)
.map(|datom| datom_edn(&db, datom))
.collect();
Ok(edn_to_sv(&Edn::Vector(items)))
});
let base = db.clone();
host.define("corium.api", "as-of", move |args: &[SerializedValue]| {
arity(args, 2, "as-of")?;
let mut view =
token_view(&args[0]).ok_or("as-of takes a database as its first argument")?;
view.as_of = Some(t_arg(&args[1])?);
Ok(view.token(&base))
});
let base = db.clone();
host.define("corium.api", "since", move |args: &[SerializedValue]| {
arity(args, 2, "since")?;
let mut view =
token_view(&args[0]).ok_or("since takes a database as its first argument")?;
view.since = Some(t_arg(&args[1])?);
Ok(view.token(&base))
});
let base = db.clone();
host.define(
"corium.api",
"history",
move |args: &[SerializedValue]| {
arity(args, 1, "history")?;
let mut view =
token_view(&args[0]).ok_or("history takes a database as its argument")?;
view.history = true;
Ok(view.token(&base))
},
);
let base = db.clone();
host.define(
"corium.api",
"basis-t",
move |args: &[SerializedValue]| {
arity(args, 1, "basis-t")?;
token_view(&args[0]).ok_or("basis-t takes a database as its argument")?;
Ok(SerializedValue::Long(
i64::try_from(base.basis_t()).unwrap_or(i64::MAX),
))
},
);
host
}
fn t_arg(value: &SerializedValue) -> Result<u64, String> {
match value {
SerializedValue::Long(t) => u64::try_from(*t).map_err(|_| "t must be non-negative".into()),
_ => Err("t must be a long".into()),
}
}
fn eid_of(db: &Db, form: &Edn) -> Result<EntityId, String> {
match form {
Edn::Long(n) => u64::try_from(*n)
.map(EntityId::from_raw)
.map_err(|_| "entity id must be non-negative".into()),
Edn::Keyword(keyword) => db
.idents()
.entid(keyword)
.ok_or_else(|| format!("unknown ident {keyword}")),
Edn::Tagged(tag, inner) if tag == "eid" => match inner.as_ref() {
Edn::Long(n) => u64::try_from(*n)
.map(EntityId::from_raw)
.map_err(|_| "entity id must be non-negative".into()),
_ => Err("#eid requires a long".into()),
},
Edn::Vector(items) => {
let [attr_form, value_form] = items.as_slice() else {
return Err("lookup ref must be [attr value]".into());
};
let attr = attr_form
.as_keyword()
.and_then(|keyword| db.idents().entid(keyword))
.ok_or_else(|| format!("unknown lookup attribute {attr_form}"))?;
let value_type = db
.schema()
.get(attr)
.map(|meta| meta.value_type)
.ok_or("lookup attribute is not installed")?;
let value = edn_to_value(Some(db), value_form)
.map(|value| corium_query::exec::coerce_for_type(value, value_type))
.ok_or_else(|| format!("bad lookup value {value_form}"))?;
db.lookup(attr, &value)
.ok_or_else(|| format!("lookup ref [{attr_form} {value_form}] not found"))
}
other => Err(format!("bad entity position {other}")),
}
}
fn datom_edn(db: &Db, datom: &corium_core::Datom) -> Edn {
let attr = db.idents().ident(datom.a).map_or_else(
|| Edn::Long(i64::try_from(datom.a.raw()).unwrap_or(i64::MAX)),
|keyword| Edn::Keyword(keyword.clone()),
);
Edn::Vector(vec![
Edn::Long(i64::try_from(datom.e.raw()).unwrap_or(i64::MAX)),
attr,
value_to_edn(db, &datom.v),
Edn::Long(i64::try_from(datom.tx.raw()).unwrap_or(i64::MAX)),
Edn::Bool(datom.added),
])
}
const TAG_KEY: &str = "tag";
fn tag_meta_key() -> SerializedValue {
SerializedValue::Keyword {
namespace: Some("corium".into()),
name: TAG_KEY.into(),
}
}
fn edn_to_sv(form: &Edn) -> SerializedValue {
match form {
Edn::Nil => SerializedValue::Nil,
Edn::Bool(v) => SerializedValue::Bool(*v),
Edn::Long(v) => SerializedValue::Long(*v),
Edn::Double(v) => SerializedValue::Double(v.0),
Edn::Str(v) => SerializedValue::Str(v.clone()),
Edn::Keyword(k) => SerializedValue::Keyword {
namespace: k.namespace.as_deref().map(Arc::from),
name: Arc::from(k.name.as_str()),
},
Edn::Symbol(s) => {
let (namespace, name) = match s.split_once('/') {
Some((namespace, name)) if !namespace.is_empty() && !name.is_empty() => {
(Some(Arc::from(namespace)), Arc::from(name))
}
_ => (None, Arc::from(s.as_str())),
};
SerializedValue::Symbol {
namespace,
name,
version: None,
}
}
Edn::List(items) => SerializedValue::List(items.iter().map(edn_to_sv).collect()),
Edn::Vector(items) => SerializedValue::Vector(items.iter().map(edn_to_sv).collect()),
Edn::Map(pairs) => SerializedValue::ArrayMap(
pairs
.iter()
.map(|(k, v)| (edn_to_sv(k), edn_to_sv(v)))
.collect(),
),
Edn::Set(items) => SerializedValue::HashSet(items.iter().map(edn_to_sv).collect()),
Edn::Tagged(tag, inner) => match (tag.as_str(), inner.as_ref()) {
("uuid", Edn::Str(hex)) => u128::from_str_radix(hex, 16)
.map_or_else(|_| tagged_fallback(tag, inner), SerializedValue::Uuid),
_ => tagged_fallback(tag, inner),
},
}
}
fn tagged_fallback(tag: &str, inner: &Edn) -> SerializedValue {
SerializedValue::WithMeta {
value: Box::new(edn_to_sv(inner)),
meta: Box::new(SerializedValue::ArrayMap(vec![(
tag_meta_key(),
SerializedValue::Keyword {
namespace: None,
name: Arc::from(tag),
},
)])),
}
}
fn sv_to_edn(value: &SerializedValue) -> Result<Edn, String> {
match value {
SerializedValue::Nil => Ok(Edn::Nil),
SerializedValue::Bool(v) => Ok(Edn::Bool(*v)),
SerializedValue::Long(v) => Ok(Edn::Long(*v)),
SerializedValue::Double(v) => Ok(Edn::Double(TotalF64(*v))),
SerializedValue::Char(c) => Ok(Edn::Str(c.to_string())),
SerializedValue::Str(v) => Ok(Edn::Str(v.clone())),
SerializedValue::Uuid(v) => Ok(Edn::Tagged(
"uuid".into(),
Box::new(Edn::Str(format!("{v:032x}"))),
)),
SerializedValue::Keyword { namespace, name } => Ok(Edn::Keyword(Keyword {
namespace: namespace.as_deref().map(str::to_owned),
name: name.to_string(),
})),
SerializedValue::Symbol {
namespace, name, ..
} => Ok(Edn::Symbol(namespace.as_deref().map_or_else(
|| name.to_string(),
|namespace| format!("{namespace}/{name}"),
))),
SerializedValue::List(items) | SerializedValue::Queue(items) => Ok(Edn::List(
items.iter().map(sv_to_edn).collect::<Result<_, _>>()?,
)),
SerializedValue::Vector(items) => Ok(Edn::Vector(
items.iter().map(sv_to_edn).collect::<Result<_, _>>()?,
)),
SerializedValue::Cons { .. } => {
let mut items = Vec::new();
let mut current = value;
loop {
match current {
SerializedValue::Cons { head, tail } => {
items.push(sv_to_edn(head)?);
current = tail;
}
SerializedValue::Nil => break,
SerializedValue::List(rest) | SerializedValue::Vector(rest) => {
for item in rest {
items.push(sv_to_edn(item)?);
}
break;
}
other => {
items.push(sv_to_edn(other)?);
break;
}
}
}
Ok(Edn::List(items))
}
SerializedValue::ArrayMap(pairs)
| SerializedValue::HashMap(pairs)
| SerializedValue::SortedMap(pairs) => {
let mut converted = pairs
.iter()
.map(|(k, v)| Ok((sv_to_edn(k)?, sv_to_edn(v)?)))
.collect::<Result<Vec<_>, String>>()?;
converted.sort_by(|a, b| a.0.cmp(&b.0));
Ok(Edn::Map(converted))
}
SerializedValue::HashSet(items) | SerializedValue::SortedSet(items) => {
let mut converted = items.iter().map(sv_to_edn).collect::<Result<Vec<_>, _>>()?;
converted.sort();
converted.dedup();
Ok(Edn::Set(converted))
}
SerializedValue::WithMeta { value, meta } => {
let converted = sv_to_edn(value)?;
Ok(match meta_tag(meta) {
Some(tag) => Edn::Tagged(tag, Box::new(converted)),
None => converted,
})
}
other => Err(format!("value has no Corium representation: {other:?}")),
}
}
fn meta_tag(meta: &SerializedValue) -> Option<String> {
let (SerializedValue::ArrayMap(pairs)
| SerializedValue::HashMap(pairs)
| SerializedValue::SortedMap(pairs)) = meta
else {
return None;
};
pairs.iter().find_map(|(key, value)| {
if !is_keyword(key, Some("corium"), TAG_KEY) {
return None;
}
match value {
SerializedValue::Keyword { namespace, name } => Some(namespace.as_deref().map_or_else(
|| name.to_string(),
|namespace| format!("{namespace}/{name}"),
)),
_ => None,
}
})
}