#![forbid(unsafe_code)]
use std::{
collections::{BTreeMap, BTreeSet},
fmt,
};
use kcode_kweb_db::{
Error as KwebError, KwebDb, Node, NodeData, NodeId, Owner, Provenance, TransactionId,
};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum ErrorKind {
InvalidInput,
NotFound,
StaleRevision,
Unavailable,
Internal,
}
#[derive(Debug)]
pub struct Error {
kind: ErrorKind,
message: String,
}
impl Error {
pub fn kind(&self) -> ErrorKind {
self.kind
}
fn new(kind: ErrorKind, message: impl Into<String>) -> Self {
Self {
kind,
message: message.into(),
}
}
}
impl fmt::Display for Error {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for Error {}
impl From<KwebError> for Error {
fn from(error: KwebError) -> Self {
let kind = match error {
KwebError::InvalidInput(_) | KwebError::InvalidTransaction(_) => {
ErrorKind::InvalidInput
}
KwebError::NotFound(_) => ErrorKind::NotFound,
KwebError::Busy(_) => ErrorKind::Unavailable,
_ => ErrorKind::Internal,
};
Self::new(kind, error.to_string())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OwnerSelection {
Unowned,
SelfNode,
Node(NodeId),
}
impl From<OwnerSelection> for Owner {
fn from(value: OwnerSelection) -> Self {
match value {
OwnerSelection::Unowned => Owner::Unowned,
OwnerSelection::SelfNode => Owner::SelfNode,
OwnerSelection::Node(id) => Owner::Node(id),
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum Operation {
ConnectNodes(Vec<NodeId>),
ConsolidateFanout {
parent: NodeId,
fanout: Vec<NodeId>,
aggregator: NodeId,
},
SetFixedConnection {
parent: NodeId,
child: Option<NodeId>,
slot: usize,
},
CreateNode {
parents: Vec<NodeId>,
owner: OwnerSelection,
short_name: String,
short_description: String,
long_description: String,
},
UpdateNode {
id: NodeId,
owner: OwnerSelection,
short_name: String,
short_description: String,
long_description: String,
},
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Request {
pub operation: Operation,
pub expected_revisions: BTreeMap<NodeId, TransactionId>,
pub provenance: Provenance,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Outcome {
pub transaction_id: TransactionId,
pub created_node_id: Option<NodeId>,
pub affected_node_ids: Vec<NodeId>,
}
pub fn apply(database: &KwebDb, request: Request) -> Result<Outcome, Error> {
let affected = affected_nodes(&request.operation);
let expected = request
.expected_revisions
.keys()
.copied()
.collect::<BTreeSet<_>>();
if affected != expected {
return Err(Error::new(
ErrorKind::InvalidInput,
"expected_revisions must contain exactly every mutated existing node",
));
}
let mut transaction = database.start_transaction(request.provenance)?;
let referenced = referenced_nodes(&request.operation);
let mut nodes = BTreeMap::new();
for id in referenced {
nodes.insert(id, database.get_node(id)?);
}
for id in &affected {
let history = database.get_node_history(*id)?;
let visible = history.visible.ok_or_else(|| {
Error::new(
ErrorKind::NotFound,
format!("node {id} has no visible transaction"),
)
})?;
if request.expected_revisions.get(id) != Some(&visible) {
return Err(Error::new(
ErrorKind::StaleRevision,
format!("node {id} changed after the command was prepared"),
));
}
}
let mut updates = BTreeMap::<NodeId, NodeData>::new();
let mut created = None;
match request.operation {
Operation::ConnectNodes(ids) => {
if ids.len() < 2 {
return Err(Error::new(
ErrorKind::InvalidInput,
"ConnectNodes needs at least two nodes",
));
}
for id in &ids {
let mut data = data(&nodes, *id)?;
let mut recent = ids
.iter()
.copied()
.filter(|other| other != id)
.collect::<Vec<_>>();
for other in data.recent_connections.drain(..) {
if other != *id && !recent.contains(&other) {
recent.push(other);
}
}
data.recent_connections = recent;
updates.insert(*id, data);
}
}
Operation::ConsolidateFanout {
parent,
fanout,
aggregator,
} => {
let mut parent_data = data(&nodes, parent)?;
parent_data
.recent_connections
.retain(|id| !fanout.contains(id));
if !parent_data.recent_connections.contains(&aggregator) {
parent_data.recent_connections.push(aggregator);
}
updates.insert(parent, parent_data);
let mut aggregator_data = data(&nodes, aggregator)?;
for id in fanout {
if !aggregator_data.recent_connections.contains(&id) {
aggregator_data.recent_connections.push(id);
}
}
updates.insert(aggregator, aggregator_data);
}
Operation::SetFixedConnection {
parent,
child,
slot,
} => {
if slot == 0 {
return Err(Error::new(
ErrorKind::InvalidInput,
"fixed slots are one-based",
));
}
if child == Some(parent) {
return Err(Error::new(
ErrorKind::InvalidInput,
"a node cannot connect to itself",
));
}
let mut parent_data = data(&nodes, parent)?;
if let Some(child) = child {
if slot > parent_data.fixed_connections.len() + 1 {
return Err(Error::new(
ErrorKind::InvalidInput,
"fixed connection positions must remain contiguous",
));
}
parent_data.fixed_connections.retain(|id| *id != child);
if slot - 1 < parent_data.fixed_connections.len() {
parent_data.fixed_connections[slot - 1] = child;
} else {
parent_data.fixed_connections.push(child);
}
} else if slot - 1 < parent_data.fixed_connections.len() {
parent_data.fixed_connections.remove(slot - 1);
}
updates.insert(parent, parent_data);
}
Operation::CreateNode {
parents,
owner,
short_name,
short_description,
long_description,
} => {
if parents.is_empty() {
return Err(Error::new(
ErrorKind::InvalidInput,
"CreateNode needs at least one parent",
));
}
let id = transaction.create_node(NodeData {
short_name,
short_description,
long_description,
owner: owner.into(),
fixed_connections: Vec::new(),
recent_connections: parents.clone(),
objects: Vec::new(),
})?;
for parent in parents {
let mut parent_data = data(&nodes, parent)?;
parent_data
.recent_connections
.retain(|candidate| *candidate != id);
parent_data.recent_connections.insert(0, id);
updates.insert(parent, parent_data);
}
created = Some(id);
}
Operation::UpdateNode {
id,
owner,
short_name,
short_description,
long_description,
} => {
let mut node = data(&nodes, id)?;
node.owner = owner.into();
node.short_name = short_name;
node.short_description = short_description;
node.long_description = long_description;
updates.insert(id, node);
}
}
for (id, node) in updates {
transaction.update_node(id, node)?;
}
let transaction_id = transaction.finalize()?;
let mut affected_node_ids = affected.into_iter().collect::<Vec<_>>();
if let Some(id) = created {
affected_node_ids.push(id);
}
Ok(Outcome {
transaction_id,
created_node_id: created,
affected_node_ids,
})
}
fn data(nodes: &BTreeMap<NodeId, Node>, id: NodeId) -> Result<NodeData, Error> {
nodes
.get(&id)
.map(|node| node.data.clone())
.ok_or_else(|| Error::new(ErrorKind::NotFound, format!("node {id}")))
}
fn affected_nodes(operation: &Operation) -> BTreeSet<NodeId> {
match operation {
Operation::ConnectNodes(ids) => ids.iter().copied().collect(),
Operation::ConsolidateFanout {
parent, aggregator, ..
} => [*parent, *aggregator].into_iter().collect(),
Operation::SetFixedConnection { parent, .. } => [*parent].into_iter().collect(),
Operation::CreateNode { parents, .. } => parents.iter().copied().collect(),
Operation::UpdateNode { id, .. } => [*id].into_iter().collect(),
}
}
fn referenced_nodes(operation: &Operation) -> BTreeSet<NodeId> {
let mut ids = affected_nodes(operation);
match operation {
Operation::ConnectNodes(_) => {}
Operation::ConsolidateFanout { fanout, .. } => ids.extend(fanout),
Operation::SetFixedConnection { child, .. } => ids.extend(child),
Operation::CreateNode {
owner: OwnerSelection::Node(owner),
..
}
| Operation::UpdateNode {
owner: OwnerSelection::Node(owner),
..
} => {
ids.insert(*owner);
}
_ => {}
}
ids
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::{TimeZone, Utc};
use kcode_kweb_db::{Config, NoopGossip, WriterId};
use std::sync::Arc;
fn provenance(label: &str) -> Provenance {
Provenance {
author: "admin".into(),
source: "test".into(),
source_created_at: Utc.with_ymd_and_hms(2026, 8, 5, 0, 0, 0).unwrap(),
data: label.into(),
}
}
fn database() -> (tempfile::TempDir, KwebDb, Vec<NodeId>, TransactionId) {
let directory = tempfile::tempdir().unwrap();
let key = [3; 32];
let db = KwebDb::open(
directory.path(),
Config {
signing_key: key,
writers_by_priority: vec![WriterId::from_signing_key(&key)],
gossip: Arc::new(NoopGossip),
},
)
.unwrap();
let mut tx = db.start_transaction(provenance("seed")).unwrap();
let ids = (0..3)
.map(|index| {
tx.create_node(NodeData {
short_name: format!("Node {index}"),
short_description: String::new(),
long_description: String::new(),
owner: Owner::SelfNode,
fixed_connections: Vec::new(),
recent_connections: Vec::new(),
objects: Vec::new(),
})
.unwrap()
})
.collect::<Vec<_>>();
let revision = tx.finalize().unwrap();
(directory, db, ids, revision)
}
#[test]
fn connects_nodes_in_supplied_order_and_rejects_stale_revisions() {
let (_directory, db, ids, revision) = database();
let expected = ids.iter().map(|id| (*id, revision)).collect();
let outcome = apply(
&db,
Request {
operation: Operation::ConnectNodes(ids.clone()),
expected_revisions: expected,
provenance: provenance("connect"),
},
)
.unwrap();
assert_eq!(
db.get_node(ids[0]).unwrap().data.recent_connections,
vec![ids[1], ids[2]]
);
let stale = ids.iter().map(|id| (*id, revision)).collect();
let error = apply(
&db,
Request {
operation: Operation::ConnectNodes(ids),
expected_revisions: stale,
provenance: provenance("stale"),
},
)
.unwrap_err();
assert_eq!(error.kind(), ErrorKind::StaleRevision);
assert_ne!(outcome.transaction_id, revision);
}
#[test]
fn creates_one_node_and_updates_every_parent_in_one_transaction() {
let (_directory, db, ids, revision) = database();
let expected = ids[..2].iter().map(|id| (*id, revision)).collect();
let outcome = apply(
&db,
Request {
operation: Operation::CreateNode {
parents: ids[..2].to_vec(),
owner: OwnerSelection::SelfNode,
short_name: "Created".into(),
short_description: "short".into(),
long_description: "long".into(),
},
expected_revisions: expected,
provenance: provenance("create"),
},
)
.unwrap();
let created = outcome.created_node_id.unwrap();
assert_eq!(
db.get_node(created).unwrap().data.recent_connections,
ids[..2]
);
for parent in &ids[..2] {
let history = db.get_node_history(*parent).unwrap();
assert_eq!(history.visible, Some(outcome.transaction_id));
assert_eq!(
db.get_node(*parent).unwrap().data.recent_connections[0],
created
);
}
}
}