1use std::collections::HashMap;
20use std::collections::hash_map::DefaultHasher;
21use std::hash::{Hash, Hasher};
22use std::sync::{Arc, Mutex};
23
24use cljrs_tx::{HostApi, TxError, TxLimits, TxProgram};
25use cljrs_value::clone::SerializedValue;
26use corium_core::{EntityId, IndexOrder, Keyword, TotalF64, Value as CoreValue};
27use corium_db::Db;
28use corium_query::QInput;
29use corium_query::boundary::{edn_to_value, value_to_edn};
30use corium_query::edn::Edn;
31use thiserror::Error;
32
33use crate::node::TxFnExpander;
34
35#[derive(Clone, Copy, Debug)]
37pub struct DbFnBudget {
38 pub fuel: u64,
40 pub memory_bytes: usize,
42 pub call_depth: u64,
46}
47
48impl Default for DbFnBudget {
49 fn default() -> Self {
50 let defaults = TxLimits::default();
51 Self {
52 fuel: defaults.gas,
53 memory_bytes: defaults.memory_bytes,
54 call_depth: defaults.call_depth,
55 }
56 }
57}
58
59pub const INVOCATION_STACK_BYTES: usize = 64 * 1024 * 1024;
62
63#[derive(Debug, Error)]
65pub enum DbFnError {
66 #[error("db function failed: {0}")]
68 Execution(#[from] TxError),
69 #[error("db function recursion limit ({0}) exceeded")]
71 Recursion(usize),
72 #[error("db function must return a sequence of tx forms, got {0}")]
74 BadResult(String),
75 #[error("db function boundary conversion failed: {0}")]
77 Convert(String),
78}
79
80const NATIVE_OPS: &[&str] = &["db/add", "db/retract", "db/cas", "db/retractEntity"];
82
83pub struct DbFnExpander {
85 budget: DbFnBudget,
86 max_depth: usize,
87 programs: Mutex<HashMap<u64, Arc<TxProgram>>>,
90}
91
92impl Default for DbFnExpander {
93 fn default() -> Self {
94 Self::new(DbFnBudget::default())
95 }
96}
97
98impl DbFnExpander {
99 #[must_use]
102 pub fn new(budget: DbFnBudget) -> Self {
103 Self {
104 budget,
105 max_depth: 16,
106 programs: Mutex::new(HashMap::new()),
107 }
108 }
109
110 #[must_use]
112 pub fn with_max_depth(mut self, max_depth: usize) -> Self {
113 self.max_depth = max_depth;
114 self
115 }
116
117 pub fn expand(&self, db: &Db, forms: Vec<Edn>) -> Result<Vec<Edn>, DbFnError> {
124 let host = read_api(db);
125 self.expand_at(db, &host, forms, self.max_depth)
126 }
127
128 fn expand_at(
129 &self,
130 db: &Db,
131 host: &HostApi,
132 forms: Vec<Edn>,
133 depth: usize,
134 ) -> Result<Vec<Edn>, DbFnError> {
135 let mut out = Vec::with_capacity(forms.len());
136 for form in forms {
137 match invocation(db, &form) {
138 None => out.push(form),
139 Some((source, args)) => {
140 if depth == 0 {
141 return Err(DbFnError::Recursion(self.max_depth));
142 }
143 let produced = self.invoke(db, host, &source, &args)?;
144 out.extend(self.expand_at(db, host, produced, depth - 1)?);
145 }
146 }
147 }
148 Ok(out)
149 }
150
151 fn invoke(
153 &self,
154 db: &Db,
155 host: &HostApi,
156 source: &str,
157 args: &[Edn],
158 ) -> Result<Vec<Edn>, DbFnError> {
159 let program = self.program(source)?;
160 let mut call_args = Vec::with_capacity(args.len() + 1);
161 call_args.push(db_token(db));
162 for arg in args {
163 call_args.push(edn_to_sv(arg));
164 }
165 let limits = TxLimits {
166 memory_bytes: self.budget.memory_bytes,
167 gas: self.budget.fuel,
168 call_depth: self.budget.call_depth,
169 };
170 let result = std::thread::scope(|scope| {
174 std::thread::Builder::new()
175 .name("corium-txfn".into())
176 .stack_size(INVOCATION_STACK_BYTES)
177 .spawn_scoped(scope, || {
178 cljrs_tx::execute_with_host(&program, call_args, limits, host)
179 })
180 .map_err(|error| {
181 DbFnError::Convert(format!("cannot spawn invocation thread: {error}"))
182 })?
183 .join()
184 .map_err(|_| DbFnError::Convert("invocation thread panicked".into()))?
185 .map_err(DbFnError::from)
186 })?;
187 let items = match &result {
188 SerializedValue::List(items) | SerializedValue::Vector(items) => items.clone(),
189 SerializedValue::Nil => Vec::new(),
190 other => return Err(DbFnError::BadResult(format!("{other:?}"))),
191 };
192 items
193 .iter()
194 .map(|item| sv_to_edn(item).map_err(DbFnError::Convert))
195 .collect()
196 }
197
198 fn program(&self, source: &str) -> Result<Arc<TxProgram>, DbFnError> {
200 let mut hasher = DefaultHasher::new();
201 source.hash(&mut hasher);
202 let key = hasher.finish();
203 let mut programs = self
204 .programs
205 .lock()
206 .unwrap_or_else(std::sync::PoisonError::into_inner);
207 if let Some(program) = programs.get(&key) {
208 return Ok(Arc::clone(program));
209 }
210 let program = Arc::new(TxProgram::parse(source)?);
211 programs.insert(key, Arc::clone(&program));
212 Ok(program)
213 }
214}
215
216impl TxFnExpander for DbFnExpander {
217 fn expand(&self, db: &Db, forms: Vec<Edn>) -> Result<Vec<Edn>, String> {
218 Self::expand(self, db, forms).map_err(|error| error.to_string())
219 }
220}
221
222fn invocation(db: &Db, form: &Edn) -> Option<(String, Vec<Edn>)> {
225 let (Edn::Vector(items) | Edn::List(items)) = form else {
226 return None;
227 };
228 let head = items.first()?.as_keyword()?;
229 let full = head.namespace.as_deref().map_or_else(
230 || head.name.clone(),
231 |namespace| format!("{namespace}/{}", head.name),
232 );
233 if NATIVE_OPS.contains(&full.as_str()) {
234 return None;
235 }
236 let source = fn_source(db, head)?;
237 Some((source, items[1..].to_vec()))
238}
239
240fn fn_source(db: &Db, ident: &Keyword) -> Option<String> {
244 let fn_attr = db.idents().entid(&Keyword::parse("db/fn"))?;
245 let entity = db.idents().entid(ident).or_else(|| {
246 let ident_attr = db.idents().entid(&Keyword::parse("db/ident"))?;
247 let interned = db.interner().get(ident)?;
248 db.lookup(ident_attr, &CoreValue::Keyword(interned))
249 })?;
250 match db.values(entity, fn_attr).into_iter().next() {
251 Some(CoreValue::Str(source)) => Some(source.to_string()),
252 _ => None,
253 }
254}
255
256const TOKEN_BASIS: &str = "basis-t";
264const TOKEN_AS_OF: &str = "as-of";
265const TOKEN_SINCE: &str = "since";
266const TOKEN_HISTORY: &str = "history";
267
268fn token_key(name: &str) -> SerializedValue {
269 SerializedValue::Keyword {
270 namespace: Some("corium.db".into()),
271 name: name.into(),
272 }
273}
274
275fn is_keyword(value: &SerializedValue, namespace: Option<&str>, name: &str) -> bool {
277 matches!(
278 value,
279 SerializedValue::Keyword {
280 namespace: kw_namespace,
281 name: kw_name,
282 } if kw_namespace.as_deref() == namespace && &**kw_name == name
283 )
284}
285
286#[derive(Clone, Copy, Default)]
288struct TokenView {
289 as_of: Option<u64>,
290 since: Option<u64>,
291 history: bool,
292}
293
294impl TokenView {
295 fn apply(self, db: &Db) -> Db {
296 let mut view = db.clone();
297 if let Some(t) = self.as_of {
298 view = view.as_of(t);
299 }
300 if let Some(t) = self.since {
301 view = view.since(t);
302 }
303 if self.history {
304 view = view.history();
305 }
306 view
307 }
308
309 fn token(self, db: &Db) -> SerializedValue {
310 let optional_t = |t: Option<u64>| {
311 t.map_or(SerializedValue::Nil, |t| {
312 SerializedValue::Long(i64::try_from(t).unwrap_or(i64::MAX))
313 })
314 };
315 SerializedValue::ArrayMap(vec![
316 (
317 token_key(TOKEN_BASIS),
318 SerializedValue::Long(i64::try_from(db.basis_t()).unwrap_or(i64::MAX)),
319 ),
320 (token_key(TOKEN_AS_OF), optional_t(self.as_of)),
321 (token_key(TOKEN_SINCE), optional_t(self.since)),
322 (
323 token_key(TOKEN_HISTORY),
324 SerializedValue::Bool(self.history),
325 ),
326 ])
327 }
328}
329
330fn db_token(db: &Db) -> SerializedValue {
332 TokenView::default().token(db)
333}
334
335fn token_view(value: &SerializedValue) -> Option<TokenView> {
338 let (SerializedValue::ArrayMap(pairs)
339 | SerializedValue::HashMap(pairs)
340 | SerializedValue::SortedMap(pairs)) = value
341 else {
342 return None;
343 };
344 let field = |name: &str| {
345 pairs
346 .iter()
347 .find(|(key, _)| is_keyword(key, Some("corium.db"), name))
348 .map(|(_, value)| value)
349 };
350 field(TOKEN_BASIS)?;
353 let t_of = |name: &str| match field(name) {
354 Some(SerializedValue::Long(t)) => u64::try_from(*t).ok(),
355 _ => None,
356 };
357 Some(TokenView {
358 as_of: t_of(TOKEN_AS_OF),
359 since: t_of(TOKEN_SINCE),
360 history: matches!(field(TOKEN_HISTORY), Some(SerializedValue::Bool(true))),
361 })
362}
363
364#[allow(clippy::too_many_lines)]
368fn read_api(db: &Db) -> HostApi {
369 let mut host = HostApi::new();
370 let arity = |args: &[SerializedValue], want: usize, what: &str| {
371 if args.len() == want {
372 Ok(())
373 } else {
374 Err(format!("{what} takes {want} arguments, got {}", args.len()))
375 }
376 };
377 let view_arg = |db: &Db, value: &SerializedValue, what: &str| {
378 token_view(value)
379 .map(|view| view.apply(db))
380 .ok_or_else(|| format!("{what} takes a database as its first argument"))
381 };
382
383 let base = db.clone();
384 host.define("corium.api", "q", move |args: &[SerializedValue]| {
385 if args.len() < 2 {
386 return Err("q takes a query and at least one input".into());
387 }
388 let query = sv_to_edn(&args[0])?;
389 let inputs: Vec<Result<Db, Edn>> = args[1..]
392 .iter()
393 .map(|arg| match token_view(arg) {
394 Some(view) => Ok(Ok(view.apply(&base))),
395 None => sv_to_edn(arg).map(Err),
396 })
397 .collect::<Result<_, _>>()?;
398 let qinputs: Vec<QInput<'_>> = inputs
399 .iter()
400 .map(|input| match input {
401 Ok(db) => QInput::Db(db),
402 Err(edn) => QInput::Edn(edn.clone()),
403 })
404 .collect();
405 let result = corium_query::q(&query, &qinputs).map_err(|error| error.to_string())?;
406 Ok(edn_to_sv(&result))
407 });
408
409 let base = db.clone();
410 host.define("corium.api", "pull", move |args: &[SerializedValue]| {
411 arity(args, 3, "pull")?;
412 let db = view_arg(&base, &args[0], "pull")?;
413 let pattern = sv_to_edn(&args[1])?;
414 let eid = eid_of(&db, &sv_to_edn(&args[2])?)?;
415 let result = corium_query::pull(&db, &pattern, eid).map_err(|error| error.to_string())?;
416 Ok(edn_to_sv(&result))
417 });
418
419 let base = db.clone();
420 host.define("corium.api", "entity", move |args: &[SerializedValue]| {
421 arity(args, 2, "entity")?;
422 let db = view_arg(&base, &args[0], "entity")?;
423 let eid = eid_of(&db, &sv_to_edn(&args[1])?)?;
424 let entity = corium_query::Entity::new(&db, eid);
425 let mut pairs = vec![(
426 Edn::keyword("db/id"),
427 Edn::Long(i64::try_from(eid.raw()).unwrap_or(i64::MAX)),
428 )];
429 for attr in entity.keys() {
430 let Some(keyword) = db.idents().ident(attr) else {
431 continue;
432 };
433 let values = entity.get(attr);
434 let many = db
435 .schema()
436 .get(attr)
437 .is_some_and(|meta| meta.cardinality == corium_core::Cardinality::Many);
438 let value = if many {
439 let mut items: Vec<Edn> = values
440 .iter()
441 .map(|value| value_to_edn(&db, value))
442 .collect();
443 items.sort();
444 items.dedup();
445 Edn::Set(items)
446 } else if let Some(value) = values.first() {
447 value_to_edn(&db, value)
448 } else {
449 continue;
450 };
451 pairs.push((Edn::Keyword(keyword.clone()), value));
452 }
453 pairs.sort_by(|a, b| a.0.cmp(&b.0));
454 Ok(edn_to_sv(&Edn::Map(pairs)))
455 });
456
457 let base = db.clone();
458 host.define("corium.api", "datoms", move |args: &[SerializedValue]| {
459 arity(args, 2, "datoms")?;
460 let db = view_arg(&base, &args[0], "datoms")?;
461 let index = sv_to_edn(&args[1])?;
462 let order = match index.as_keyword().map(|keyword| keyword.name.as_str()) {
463 Some("eavt") => IndexOrder::Eavt,
464 Some("aevt") => IndexOrder::Aevt,
465 Some("avet") => IndexOrder::Avet,
466 Some("vaet") => IndexOrder::Vaet,
467 _ => return Err("index must be :eavt, :aevt, :avet, or :vaet".into()),
468 };
469 let items: Vec<Edn> = db
470 .datoms_at(order)
471 .map(|datom| datom_edn(&db, datom))
472 .collect();
473 Ok(edn_to_sv(&Edn::Vector(items)))
474 });
475
476 let base = db.clone();
477 host.define("corium.api", "as-of", move |args: &[SerializedValue]| {
478 arity(args, 2, "as-of")?;
479 let mut view =
480 token_view(&args[0]).ok_or("as-of takes a database as its first argument")?;
481 view.as_of = Some(t_arg(&args[1])?);
482 Ok(view.token(&base))
483 });
484
485 let base = db.clone();
486 host.define("corium.api", "since", move |args: &[SerializedValue]| {
487 arity(args, 2, "since")?;
488 let mut view =
489 token_view(&args[0]).ok_or("since takes a database as its first argument")?;
490 view.since = Some(t_arg(&args[1])?);
491 Ok(view.token(&base))
492 });
493
494 let base = db.clone();
495 host.define(
496 "corium.api",
497 "history",
498 move |args: &[SerializedValue]| {
499 arity(args, 1, "history")?;
500 let mut view =
501 token_view(&args[0]).ok_or("history takes a database as its argument")?;
502 view.history = true;
503 Ok(view.token(&base))
504 },
505 );
506
507 let base = db.clone();
508 host.define(
509 "corium.api",
510 "basis-t",
511 move |args: &[SerializedValue]| {
512 arity(args, 1, "basis-t")?;
513 token_view(&args[0]).ok_or("basis-t takes a database as its argument")?;
514 Ok(SerializedValue::Long(
515 i64::try_from(base.basis_t()).unwrap_or(i64::MAX),
516 ))
517 },
518 );
519
520 host
521}
522
523fn t_arg(value: &SerializedValue) -> Result<u64, String> {
524 match value {
525 SerializedValue::Long(t) => u64::try_from(*t).map_err(|_| "t must be non-negative".into()),
526 _ => Err("t must be a long".into()),
527 }
528}
529
530fn eid_of(db: &Db, form: &Edn) -> Result<EntityId, String> {
532 match form {
533 Edn::Long(n) => u64::try_from(*n)
534 .map(EntityId::from_raw)
535 .map_err(|_| "entity id must be non-negative".into()),
536 Edn::Keyword(keyword) => db
537 .idents()
538 .entid(keyword)
539 .ok_or_else(|| format!("unknown ident {keyword}")),
540 Edn::Tagged(tag, inner) if tag == "eid" => match inner.as_ref() {
541 Edn::Long(n) => u64::try_from(*n)
542 .map(EntityId::from_raw)
543 .map_err(|_| "entity id must be non-negative".into()),
544 _ => Err("#eid requires a long".into()),
545 },
546 Edn::Vector(items) => {
547 let [attr_form, value_form] = items.as_slice() else {
548 return Err("lookup ref must be [attr value]".into());
549 };
550 let attr = attr_form
551 .as_keyword()
552 .and_then(|keyword| db.idents().entid(keyword))
553 .ok_or_else(|| format!("unknown lookup attribute {attr_form}"))?;
554 let value_type = db
555 .schema()
556 .get(attr)
557 .map(|meta| meta.value_type)
558 .ok_or("lookup attribute is not installed")?;
559 let value = edn_to_value(Some(db), value_form)
560 .map(|value| corium_query::exec::coerce_for_type(value, value_type))
561 .ok_or_else(|| format!("bad lookup value {value_form}"))?;
562 db.lookup(attr, &value)
563 .ok_or_else(|| format!("lookup ref [{attr_form} {value_form}] not found"))
564 }
565 other => Err(format!("bad entity position {other}")),
566 }
567}
568
569fn datom_edn(db: &Db, datom: &corium_core::Datom) -> Edn {
570 let attr = db.idents().ident(datom.a).map_or_else(
571 || Edn::Long(i64::try_from(datom.a.raw()).unwrap_or(i64::MAX)),
572 |keyword| Edn::Keyword(keyword.clone()),
573 );
574 Edn::Vector(vec![
575 Edn::Long(i64::try_from(datom.e.raw()).unwrap_or(i64::MAX)),
576 attr,
577 value_to_edn(db, &datom.v),
578 Edn::Long(i64::try_from(datom.tx.raw()).unwrap_or(i64::MAX)),
579 Edn::Bool(datom.added),
580 ])
581}
582
583const TAG_KEY: &str = "tag";
592
593fn tag_meta_key() -> SerializedValue {
594 SerializedValue::Keyword {
595 namespace: Some("corium".into()),
596 name: TAG_KEY.into(),
597 }
598}
599
600fn edn_to_sv(form: &Edn) -> SerializedValue {
602 match form {
603 Edn::Nil => SerializedValue::Nil,
604 Edn::Bool(v) => SerializedValue::Bool(*v),
605 Edn::Long(v) => SerializedValue::Long(*v),
606 Edn::Double(v) => SerializedValue::Double(v.0),
607 Edn::Str(v) => SerializedValue::Str(v.clone()),
608 Edn::Keyword(k) => SerializedValue::Keyword {
609 namespace: k.namespace.as_deref().map(Arc::from),
610 name: Arc::from(k.name.as_str()),
611 },
612 Edn::Symbol(s) => {
613 let (namespace, name) = match s.split_once('/') {
614 Some((namespace, name)) if !namespace.is_empty() && !name.is_empty() => {
615 (Some(Arc::from(namespace)), Arc::from(name))
616 }
617 _ => (None, Arc::from(s.as_str())),
618 };
619 SerializedValue::Symbol {
620 namespace,
621 name,
622 version: None,
623 }
624 }
625 Edn::List(items) => SerializedValue::List(items.iter().map(edn_to_sv).collect()),
626 Edn::Vector(items) => SerializedValue::Vector(items.iter().map(edn_to_sv).collect()),
627 Edn::Map(pairs) => SerializedValue::ArrayMap(
628 pairs
629 .iter()
630 .map(|(k, v)| (edn_to_sv(k), edn_to_sv(v)))
631 .collect(),
632 ),
633 Edn::Set(items) => SerializedValue::HashSet(items.iter().map(edn_to_sv).collect()),
634 Edn::Tagged(tag, inner) => match (tag.as_str(), inner.as_ref()) {
635 ("uuid", Edn::Str(hex)) => u128::from_str_radix(hex, 16)
636 .map_or_else(|_| tagged_fallback(tag, inner), SerializedValue::Uuid),
637 _ => tagged_fallback(tag, inner),
638 },
639 }
640}
641
642fn tagged_fallback(tag: &str, inner: &Edn) -> SerializedValue {
645 SerializedValue::WithMeta {
646 value: Box::new(edn_to_sv(inner)),
647 meta: Box::new(SerializedValue::ArrayMap(vec![(
648 tag_meta_key(),
649 SerializedValue::Keyword {
650 namespace: None,
651 name: Arc::from(tag),
652 },
653 )])),
654 }
655}
656
657fn sv_to_edn(value: &SerializedValue) -> Result<Edn, String> {
663 match value {
664 SerializedValue::Nil => Ok(Edn::Nil),
665 SerializedValue::Bool(v) => Ok(Edn::Bool(*v)),
666 SerializedValue::Long(v) => Ok(Edn::Long(*v)),
667 SerializedValue::Double(v) => Ok(Edn::Double(TotalF64(*v))),
668 SerializedValue::Char(c) => Ok(Edn::Str(c.to_string())),
669 SerializedValue::Str(v) => Ok(Edn::Str(v.clone())),
670 SerializedValue::Uuid(v) => Ok(Edn::Tagged(
671 "uuid".into(),
672 Box::new(Edn::Str(format!("{v:032x}"))),
673 )),
674 SerializedValue::Keyword { namespace, name } => Ok(Edn::Keyword(Keyword {
675 namespace: namespace.as_deref().map(str::to_owned),
676 name: name.to_string(),
677 })),
678 SerializedValue::Symbol {
679 namespace, name, ..
680 } => Ok(Edn::Symbol(namespace.as_deref().map_or_else(
681 || name.to_string(),
682 |namespace| format!("{namespace}/{name}"),
683 ))),
684 SerializedValue::List(items) | SerializedValue::Queue(items) => Ok(Edn::List(
685 items.iter().map(sv_to_edn).collect::<Result<_, _>>()?,
686 )),
687 SerializedValue::Vector(items) => Ok(Edn::Vector(
688 items.iter().map(sv_to_edn).collect::<Result<_, _>>()?,
689 )),
690 SerializedValue::Cons { .. } => {
691 let mut items = Vec::new();
692 let mut current = value;
693 loop {
694 match current {
695 SerializedValue::Cons { head, tail } => {
696 items.push(sv_to_edn(head)?);
697 current = tail;
698 }
699 SerializedValue::Nil => break,
700 SerializedValue::List(rest) | SerializedValue::Vector(rest) => {
701 for item in rest {
702 items.push(sv_to_edn(item)?);
703 }
704 break;
705 }
706 other => {
707 items.push(sv_to_edn(other)?);
708 break;
709 }
710 }
711 }
712 Ok(Edn::List(items))
713 }
714 SerializedValue::ArrayMap(pairs)
715 | SerializedValue::HashMap(pairs)
716 | SerializedValue::SortedMap(pairs) => {
717 let mut converted = pairs
718 .iter()
719 .map(|(k, v)| Ok((sv_to_edn(k)?, sv_to_edn(v)?)))
720 .collect::<Result<Vec<_>, String>>()?;
721 converted.sort_by(|a, b| a.0.cmp(&b.0));
722 Ok(Edn::Map(converted))
723 }
724 SerializedValue::HashSet(items) | SerializedValue::SortedSet(items) => {
725 let mut converted = items.iter().map(sv_to_edn).collect::<Result<Vec<_>, _>>()?;
726 converted.sort();
727 converted.dedup();
728 Ok(Edn::Set(converted))
729 }
730 SerializedValue::WithMeta { value, meta } => {
731 let converted = sv_to_edn(value)?;
732 Ok(match meta_tag(meta) {
733 Some(tag) => Edn::Tagged(tag, Box::new(converted)),
734 None => converted,
735 })
736 }
737 other => Err(format!("value has no Corium representation: {other:?}")),
738 }
739}
740
741fn meta_tag(meta: &SerializedValue) -> Option<String> {
743 let (SerializedValue::ArrayMap(pairs)
744 | SerializedValue::HashMap(pairs)
745 | SerializedValue::SortedMap(pairs)) = meta
746 else {
747 return None;
748 };
749 pairs.iter().find_map(|(key, value)| {
750 if !is_keyword(key, Some("corium"), TAG_KEY) {
751 return None;
752 }
753 match value {
754 SerializedValue::Keyword { namespace, name } => Some(namespace.as_deref().map_or_else(
755 || name.to_string(),
756 |namespace| format!("{namespace}/{name}"),
757 )),
758 _ => None,
759 }
760 })
761}