use std::collections::{BTreeMap, BTreeSet};
use corium_core::{Cardinality, Datom, EntityId, Partition, Unique, Value};
use corium_db::{Db, FIRST_USER_ID};
use thiserror::Error;
pub type TempId = String;
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum EntityRef {
Id(EntityId),
Temp(TempId),
Lookup(EntityId, Value),
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TxOp {
Add(EntityRef, EntityId, Value),
Retract(EntityRef, EntityId, Value),
Cas(EntityRef, EntityId, Option<Value>, Value),
RetractEntity(EntityRef),
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct EntityMap {
pub entity: EntityRef,
pub attributes: Vec<(EntityId, Vec<Value>)>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TxItem {
Op(TxOp),
Map(EntityMap),
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PreparedTx {
pub datoms: Vec<Datom>,
pub tempids: BTreeMap<TempId, EntityId>,
}
#[derive(Debug, Error, Eq, PartialEq)]
pub enum TxError {
#[error("unknown attribute {0:?}")]
UnknownAttribute(EntityId),
#[error("value has wrong type for attribute {0:?}")]
TypeMismatch(EntityId),
#[error("lookup ref did not resolve")]
LookupNotFound,
#[error("lookup attribute is not unique")]
LookupNotUnique,
#[error("unique value conflict")]
UniqueConflict,
#[error("compare-and-swap failed")]
CasFailed,
#[error("compare-and-swap requires cardinality one")]
CasCardinality,
}
#[allow(clippy::too_many_lines)]
pub fn prepare(
db: &Db,
items: impl IntoIterator<Item = TxItem>,
tx: EntityId,
next_user_sequence: u64,
) -> Result<PreparedTx, TxError> {
let mut ops = Vec::new();
for item in items {
match item {
TxItem::Op(op) => ops.push(op),
TxItem::Map(map) => {
for (a, values) in map.attributes {
for value in values {
ops.push(TxOp::Add(map.entity.clone(), a, value));
}
}
}
}
}
let mut tempids = BTreeMap::new();
for op in &ops {
if let TxOp::Add(EntityRef::Temp(temp), a, value) = op {
if db.schema().get(*a).and_then(|x| x.unique) == Some(Unique::Identity) {
if let Some(e) = db.lookup(*a, value) {
tempids.insert(temp.clone(), e);
}
}
}
}
let mut next = next_user_sequence.max(FIRST_USER_ID);
for op in &ops {
let entity = match op {
TxOp::Add(e, ..) | TxOp::Retract(e, ..) | TxOp::Cas(e, ..) | TxOp::RetractEntity(e) => {
e
}
};
if let EntityRef::Temp(temp) = entity {
tempids.entry(temp.clone()).or_insert_with(|| {
let e = EntityId::new(Partition::User as u32, next);
next += 1;
e
});
}
}
let resolve = |entity: &EntityRef| -> Result<EntityId, TxError> {
match entity {
EntityRef::Id(e) => Ok(*e),
EntityRef::Temp(t) => Ok(tempids[t]),
EntityRef::Lookup(a, v) => {
let attr = db.schema().get(*a).ok_or(TxError::UnknownAttribute(*a))?;
if attr.unique.is_none() {
return Err(TxError::LookupNotUnique);
}
db.lookup(*a, v).ok_or(TxError::LookupNotFound)
}
}
};
let mut datoms = Vec::new();
let mut working = db.clone();
for op in ops {
let start = datoms.len();
match op {
TxOp::Add(entity, a, v) => {
let e = resolve(&entity)?;
validate(&working, a, &v)?;
if let Some(attr) = working.schema().get(a) {
if attr.unique.is_some()
&& working.lookup(a, &v).is_some_and(|owner| owner != e)
{
return Err(TxError::UniqueConflict);
}
let current = working.values(e, a);
if current.contains(&v) {
continue;
}
if attr.cardinality == Cardinality::One {
for old in current {
datoms.push(Datom {
e,
a,
v: old,
tx,
added: false,
});
}
}
}
datoms.push(Datom {
e,
a,
v,
tx,
added: true,
});
}
TxOp::Retract(entity, a, v) => {
validate(&working, a, &v)?;
let e = resolve(&entity)?;
if working.values(e, a).contains(&v) {
datoms.push(Datom {
e,
a,
v,
tx,
added: false,
});
}
}
TxOp::Cas(entity, a, old, new) => {
validate(&working, a, &new)?;
let e = resolve(&entity)?;
if working
.schema()
.get(a)
.is_none_or(|x| x.cardinality != Cardinality::One)
{
return Err(TxError::CasCardinality);
}
let current = working.values(e, a).into_iter().next();
if current != old {
return Err(TxError::CasFailed);
}
if let Some(value) = current {
datoms.push(Datom {
e,
a,
v: value,
tx,
added: false,
});
}
datoms.push(Datom {
e,
a,
v: new,
tx,
added: true,
});
}
TxOp::RetractEntity(entity) => {
let mut facts = BTreeSet::new();
collect_entity_retractions(
&working,
resolve(&entity)?,
&mut facts,
&mut BTreeSet::new(),
);
datoms.extend(facts.into_iter().map(|(e, a, v)| Datom {
e,
a,
v,
tx,
added: false,
}));
}
}
working = working.with_transaction(working.basis_t() + 1, &datoms[start..]);
}
Ok(PreparedTx { datoms, tempids })
}
fn validate(db: &Db, a: EntityId, value: &Value) -> Result<(), TxError> {
let attr = db.schema().get(a).ok_or(TxError::UnknownAttribute(a))?;
if !value.has_type(attr.value_type) {
return Err(TxError::TypeMismatch(a));
}
Ok(())
}
fn collect_entity_retractions(
db: &Db,
e: EntityId,
facts: &mut BTreeSet<(EntityId, EntityId, Value)>,
seen: &mut BTreeSet<EntityId>,
) {
if !seen.insert(e) {
return;
}
for datom in db.datoms() {
if datom.e == e {
if db.schema().get(datom.a).is_some_and(|a| a.is_component) {
if let Value::Ref(child) = &datom.v {
collect_entity_retractions(db, *child, facts, seen);
}
}
facts.insert((datom.e, datom.a, datom.v));
} else if datom.v == Value::Ref(e) {
facts.insert((datom.e, datom.a, datom.v));
}
}
}