1#![forbid(unsafe_code)]
2
3use std::{
4 collections::{BTreeMap, BTreeSet},
5 fmt,
6};
7
8use kcode_kweb_db::{
9 Error as KwebError, KwebDb, Node, NodeData, NodeId, Owner, Provenance, TransactionId,
10};
11
12#[derive(Clone, Copy, Debug, Eq, PartialEq)]
13#[non_exhaustive]
14pub enum ErrorKind {
15 InvalidInput,
16 NotFound,
17 StaleRevision,
18 Unavailable,
19 Internal,
20}
21
22#[derive(Debug)]
23pub struct Error {
24 kind: ErrorKind,
25 message: String,
26}
27
28impl Error {
29 pub fn kind(&self) -> ErrorKind {
30 self.kind
31 }
32 fn new(kind: ErrorKind, message: impl Into<String>) -> Self {
33 Self {
34 kind,
35 message: message.into(),
36 }
37 }
38}
39impl fmt::Display for Error {
40 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
41 formatter.write_str(&self.message)
42 }
43}
44impl std::error::Error for Error {}
45
46impl From<KwebError> for Error {
47 fn from(error: KwebError) -> Self {
48 let kind = match error {
49 KwebError::InvalidInput(_) | KwebError::InvalidTransaction(_) => {
50 ErrorKind::InvalidInput
51 }
52 KwebError::NotFound(_) => ErrorKind::NotFound,
53 KwebError::Busy(_) => ErrorKind::Unavailable,
54 _ => ErrorKind::Internal,
55 };
56 Self::new(kind, error.to_string())
57 }
58}
59
60#[derive(Clone, Copy, Debug, Eq, PartialEq)]
61pub enum OwnerSelection {
62 Unowned,
63 SelfNode,
64 Node(NodeId),
65}
66
67impl From<OwnerSelection> for Owner {
68 fn from(value: OwnerSelection) -> Self {
69 match value {
70 OwnerSelection::Unowned => Owner::Unowned,
71 OwnerSelection::SelfNode => Owner::SelfNode,
72 OwnerSelection::Node(id) => Owner::Node(id),
73 }
74 }
75}
76
77#[derive(Clone, Debug, Eq, PartialEq)]
78pub enum Operation {
79 ConnectNodes(Vec<NodeId>),
80 ConsolidateFanout {
81 parent: NodeId,
82 fanout: Vec<NodeId>,
83 aggregator: NodeId,
84 },
85 SetFixedConnection {
86 parent: NodeId,
87 child: Option<NodeId>,
88 slot: usize,
89 },
90 CreateNode {
91 parents: Vec<NodeId>,
92 owner: OwnerSelection,
93 short_name: String,
94 short_description: String,
95 long_description: String,
96 },
97 UpdateNode {
98 id: NodeId,
99 owner: OwnerSelection,
100 short_name: String,
101 short_description: String,
102 long_description: String,
103 },
104}
105
106#[derive(Clone, Debug, Eq, PartialEq)]
107pub struct Request {
108 pub operation: Operation,
109 pub expected_revisions: BTreeMap<NodeId, TransactionId>,
110 pub provenance: Provenance,
111}
112
113#[derive(Clone, Debug, Eq, PartialEq)]
114pub struct Outcome {
115 pub transaction_id: TransactionId,
116 pub created_node_id: Option<NodeId>,
117 pub affected_node_ids: Vec<NodeId>,
118}
119
120pub fn apply(database: &KwebDb, request: Request) -> Result<Outcome, Error> {
121 let affected = affected_nodes(&request.operation);
122 let expected = request
123 .expected_revisions
124 .keys()
125 .copied()
126 .collect::<BTreeSet<_>>();
127 if affected != expected {
128 return Err(Error::new(
129 ErrorKind::InvalidInput,
130 "expected_revisions must contain exactly every mutated existing node",
131 ));
132 }
133 let mut transaction = database.start_transaction(request.provenance)?;
134 let referenced = referenced_nodes(&request.operation);
135 let mut nodes = BTreeMap::new();
136 for id in referenced {
137 nodes.insert(id, database.get_node(id)?);
138 }
139 for id in &affected {
140 let history = database.get_node_history(*id)?;
141 let visible = history.visible.ok_or_else(|| {
142 Error::new(
143 ErrorKind::NotFound,
144 format!("node {id} has no visible transaction"),
145 )
146 })?;
147 if request.expected_revisions.get(id) != Some(&visible) {
148 return Err(Error::new(
149 ErrorKind::StaleRevision,
150 format!("node {id} changed after the command was prepared"),
151 ));
152 }
153 }
154
155 let mut updates = BTreeMap::<NodeId, NodeData>::new();
156 let mut created = None;
157 match request.operation {
158 Operation::ConnectNodes(ids) => {
159 if ids.len() < 2 {
160 return Err(Error::new(
161 ErrorKind::InvalidInput,
162 "ConnectNodes needs at least two nodes",
163 ));
164 }
165 for id in &ids {
166 let mut data = data(&nodes, *id)?;
167 let mut recent = ids
168 .iter()
169 .copied()
170 .filter(|other| other != id)
171 .collect::<Vec<_>>();
172 for other in data.recent_connections.drain(..) {
173 if other != *id && !recent.contains(&other) {
174 recent.push(other);
175 }
176 }
177 data.recent_connections = recent;
178 updates.insert(*id, data);
179 }
180 }
181 Operation::ConsolidateFanout {
182 parent,
183 fanout,
184 aggregator,
185 } => {
186 let mut parent_data = data(&nodes, parent)?;
187 parent_data
188 .recent_connections
189 .retain(|id| !fanout.contains(id));
190 if !parent_data.recent_connections.contains(&aggregator) {
191 parent_data.recent_connections.push(aggregator);
192 }
193 updates.insert(parent, parent_data);
194 let mut aggregator_data = data(&nodes, aggregator)?;
195 for id in fanout {
196 if !aggregator_data.recent_connections.contains(&id) {
197 aggregator_data.recent_connections.push(id);
198 }
199 }
200 updates.insert(aggregator, aggregator_data);
201 }
202 Operation::SetFixedConnection {
203 parent,
204 child,
205 slot,
206 } => {
207 if slot == 0 {
208 return Err(Error::new(
209 ErrorKind::InvalidInput,
210 "fixed slots are one-based",
211 ));
212 }
213 if child == Some(parent) {
214 return Err(Error::new(
215 ErrorKind::InvalidInput,
216 "a node cannot connect to itself",
217 ));
218 }
219 let mut parent_data = data(&nodes, parent)?;
220 if let Some(child) = child {
221 if slot > parent_data.fixed_connections.len() + 1 {
222 return Err(Error::new(
223 ErrorKind::InvalidInput,
224 "fixed connection positions must remain contiguous",
225 ));
226 }
227 parent_data.fixed_connections.retain(|id| *id != child);
228 if slot - 1 < parent_data.fixed_connections.len() {
229 parent_data.fixed_connections[slot - 1] = child;
230 } else {
231 parent_data.fixed_connections.push(child);
232 }
233 } else if slot - 1 < parent_data.fixed_connections.len() {
234 parent_data.fixed_connections.remove(slot - 1);
235 }
236 updates.insert(parent, parent_data);
237 }
238 Operation::CreateNode {
239 parents,
240 owner,
241 short_name,
242 short_description,
243 long_description,
244 } => {
245 if parents.is_empty() {
246 return Err(Error::new(
247 ErrorKind::InvalidInput,
248 "CreateNode needs at least one parent",
249 ));
250 }
251 let id = transaction.create_node(NodeData {
252 short_name,
253 short_description,
254 long_description,
255 owner: owner.into(),
256 fixed_connections: Vec::new(),
257 recent_connections: parents.clone(),
258 objects: Vec::new(),
259 })?;
260 for parent in parents {
261 let mut parent_data = data(&nodes, parent)?;
262 parent_data
263 .recent_connections
264 .retain(|candidate| *candidate != id);
265 parent_data.recent_connections.insert(0, id);
266 updates.insert(parent, parent_data);
267 }
268 created = Some(id);
269 }
270 Operation::UpdateNode {
271 id,
272 owner,
273 short_name,
274 short_description,
275 long_description,
276 } => {
277 let mut node = data(&nodes, id)?;
278 node.owner = owner.into();
279 node.short_name = short_name;
280 node.short_description = short_description;
281 node.long_description = long_description;
282 updates.insert(id, node);
283 }
284 }
285 for (id, node) in updates {
286 transaction.update_node(id, node)?;
287 }
288 let transaction_id = transaction.finalize()?;
289 let mut affected_node_ids = affected.into_iter().collect::<Vec<_>>();
290 if let Some(id) = created {
291 affected_node_ids.push(id);
292 }
293 Ok(Outcome {
294 transaction_id,
295 created_node_id: created,
296 affected_node_ids,
297 })
298}
299
300fn data(nodes: &BTreeMap<NodeId, Node>, id: NodeId) -> Result<NodeData, Error> {
301 nodes
302 .get(&id)
303 .map(|node| node.data.clone())
304 .ok_or_else(|| Error::new(ErrorKind::NotFound, format!("node {id}")))
305}
306
307fn affected_nodes(operation: &Operation) -> BTreeSet<NodeId> {
308 match operation {
309 Operation::ConnectNodes(ids) => ids.iter().copied().collect(),
310 Operation::ConsolidateFanout {
311 parent, aggregator, ..
312 } => [*parent, *aggregator].into_iter().collect(),
313 Operation::SetFixedConnection { parent, .. } => [*parent].into_iter().collect(),
314 Operation::CreateNode { parents, .. } => parents.iter().copied().collect(),
315 Operation::UpdateNode { id, .. } => [*id].into_iter().collect(),
316 }
317}
318
319fn referenced_nodes(operation: &Operation) -> BTreeSet<NodeId> {
320 let mut ids = affected_nodes(operation);
321 match operation {
322 Operation::ConnectNodes(_) => {}
323 Operation::ConsolidateFanout { fanout, .. } => ids.extend(fanout),
324 Operation::SetFixedConnection { child, .. } => ids.extend(child),
325 Operation::CreateNode {
326 owner: OwnerSelection::Node(owner),
327 ..
328 }
329 | Operation::UpdateNode {
330 owner: OwnerSelection::Node(owner),
331 ..
332 } => {
333 ids.insert(*owner);
334 }
335 _ => {}
336 }
337 ids
338}
339
340#[cfg(test)]
341mod tests {
342 use super::*;
343 use chrono::{TimeZone, Utc};
344 use kcode_kweb_db::{Config, NoopGossip, WriterId};
345 use std::sync::Arc;
346
347 fn provenance(label: &str) -> Provenance {
348 Provenance {
349 author: "admin".into(),
350 source: "test".into(),
351 source_created_at: Utc.with_ymd_and_hms(2026, 8, 5, 0, 0, 0).unwrap(),
352 data: label.into(),
353 }
354 }
355
356 fn database() -> (tempfile::TempDir, KwebDb, Vec<NodeId>, TransactionId) {
357 let directory = tempfile::tempdir().unwrap();
358 let key = [3; 32];
359 let db = KwebDb::open(
360 directory.path(),
361 Config {
362 signing_key: key,
363 writers_by_priority: vec![WriterId::from_signing_key(&key)],
364 gossip: Arc::new(NoopGossip),
365 },
366 )
367 .unwrap();
368 let mut tx = db.start_transaction(provenance("seed")).unwrap();
369 let ids = (0..3)
370 .map(|index| {
371 tx.create_node(NodeData {
372 short_name: format!("Node {index}"),
373 short_description: String::new(),
374 long_description: String::new(),
375 owner: Owner::SelfNode,
376 fixed_connections: Vec::new(),
377 recent_connections: Vec::new(),
378 objects: Vec::new(),
379 })
380 .unwrap()
381 })
382 .collect::<Vec<_>>();
383 let revision = tx.finalize().unwrap();
384 (directory, db, ids, revision)
385 }
386
387 #[test]
388 fn connects_nodes_in_supplied_order_and_rejects_stale_revisions() {
389 let (_directory, db, ids, revision) = database();
390 let expected = ids.iter().map(|id| (*id, revision)).collect();
391 let outcome = apply(
392 &db,
393 Request {
394 operation: Operation::ConnectNodes(ids.clone()),
395 expected_revisions: expected,
396 provenance: provenance("connect"),
397 },
398 )
399 .unwrap();
400 assert_eq!(
401 db.get_node(ids[0]).unwrap().data.recent_connections,
402 vec![ids[1], ids[2]]
403 );
404 let stale = ids.iter().map(|id| (*id, revision)).collect();
405 let error = apply(
406 &db,
407 Request {
408 operation: Operation::ConnectNodes(ids),
409 expected_revisions: stale,
410 provenance: provenance("stale"),
411 },
412 )
413 .unwrap_err();
414 assert_eq!(error.kind(), ErrorKind::StaleRevision);
415 assert_ne!(outcome.transaction_id, revision);
416 }
417
418 #[test]
419 fn creates_one_node_and_updates_every_parent_in_one_transaction() {
420 let (_directory, db, ids, revision) = database();
421 let expected = ids[..2].iter().map(|id| (*id, revision)).collect();
422 let outcome = apply(
423 &db,
424 Request {
425 operation: Operation::CreateNode {
426 parents: ids[..2].to_vec(),
427 owner: OwnerSelection::SelfNode,
428 short_name: "Created".into(),
429 short_description: "short".into(),
430 long_description: "long".into(),
431 },
432 expected_revisions: expected,
433 provenance: provenance("create"),
434 },
435 )
436 .unwrap();
437 let created = outcome.created_node_id.unwrap();
438 assert_eq!(
439 db.get_node(created).unwrap().data.recent_connections,
440 ids[..2]
441 );
442 for parent in &ids[..2] {
443 let history = db.get_node_history(*parent).unwrap();
444 assert_eq!(history.visible, Some(outcome.transaction_id));
445 assert_eq!(
446 db.get_node(*parent).unwrap().data.recent_connections[0],
447 created
448 );
449 }
450 }
451}