/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2023 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ use std::{cmp::Ordering, collections::HashMap, fmt::Debug, path::Path}; use async_std::sync::{Arc, Mutex}; use blake3; use darkfi_serial::{ deserialize, serialize, Decodable, Encodable, SerialDecodable, SerialEncodable, }; use log::{error, info}; use tinyjson::JsonValue; use crate::{ event_graph::events_queue::EventsQueuePtr, util::{ encoding::base64, file::{load_json_file, save_json_file}, time::Timestamp, }, }; use super::EventMsg; //pub type EventId = [u8; blake3::OUT_LEN]; pub type EventId = blake3::Hash; const MAX_DEPTH: u32 = 300; #[derive(SerialEncodable, SerialDecodable, Clone, Debug)] pub struct Event { pub previous_event_hash: EventId, pub action: T, pub timestamp: Timestamp, } impl Event where T: Send + Sync + Encodable + Decodable + Clone + EventMsg, { pub fn hash(&self) -> EventId { blake3::hash(&serialize(self)) } } #[derive(SerialEncodable, SerialDecodable, Clone, Debug)] struct EventNode { // Only current root has this set to None parent: Option, event: Event, children: Vec, } pub type ModelPtr = Arc>>; pub struct Model { // This is up to the application to reset or keep current_root: EventId, orphans: HashMap>, event_map: HashMap>, events_queue: EventsQueuePtr, } impl Model where T: Send + Sync + Encodable + Decodable + Clone + EventMsg + Debug, { pub fn new(events_queue: EventsQueuePtr) -> Self { let root_node = EventNode { parent: None, event: Event { previous_event_hash: blake3::hash(b""), // This is a blake3 hash of NULL action: T::new(), timestamp: Timestamp(1674512021323), }, children: Vec::new(), }; let root_node_id = root_node.event.hash(); let mut event_map = HashMap::new(); event_map.insert(root_node_id, root_node); Self { current_root: root_node_id, orphans: HashMap::new(), event_map, events_queue } } pub fn save_tree(&self, path: &Path) -> crate::Result<()> { let path = path.join("tree"); let tree = self.event_map.clone(); let ser_tree = base64::encode(&serialize(&tree)); save_json_file(&path, &JsonValue::String(ser_tree), false)?; info!("Tree is saved to disk"); Ok(()) } pub fn load_tree(&mut self, path: &Path) -> crate::Result<()> { let path = path.join("tree"); if !path.exists() { return Ok(()) } let loaded_tree_obj = load_json_file(&path)?; let loaded_tree_obj: &String = loaded_tree_obj.get::().unwrap(); let loaded_tree_bytes = base64::decode(loaded_tree_obj.as_str()).unwrap(); let dser_tree: HashMap> = deserialize(&loaded_tree_bytes)?; self.event_map = dser_tree; info!("Tree is loaded from disk"); Ok(()) } pub fn reset_root(&mut self, timestamp: Timestamp) { let root_node = EventNode { parent: None, event: Event { previous_event_hash: blake3::hash(b""), // This is a blake3 hash of NULL action: T::new(), timestamp, }, children: Vec::new(), }; let root_node_id = root_node.event.hash(); let mut event_map = HashMap::new(); event_map.insert(root_node_id, root_node); self.current_root = root_node_id; self.orphans = HashMap::new(); self.event_map = event_map; info!("reset current root to: {:?}", self.current_root); } pub fn remove_old_events(&mut self, timestamp: Timestamp) -> crate::Result<()> { let tree = self.event_map.clone(); let mut is_tree_changed = false; for (event_hash, node) in tree { if node.event.timestamp < timestamp { if self.event_map.remove(&event_hash).is_none() { continue } is_tree_changed = true; let parent = self.event_map.get_mut(&self.current_root).unwrap(); if parent.children.contains(&event_hash) { let index = parent.children.iter().position(|&n| n == event_hash).unwrap(); parent.children.remove(index); } } } if is_tree_changed { let binding = self.event_map.clone(); let min_hash = binding.iter().min_by_key(|entry| entry.1.event.timestamp.0).unwrap().0; println!("min hash: {}", min_hash); self.event_map.get_mut(min_hash).unwrap().parent = Some(self.current_root); self.event_map.get_mut(min_hash).unwrap().event.previous_event_hash = self.current_root; let parent = self.event_map.get_mut(&self.current_root).unwrap(); parent.children.push(*min_hash); } Ok(()) } pub fn get_head_hash(&self) -> EventId { self.find_head() } pub async fn add(&mut self, event: Event) { self.orphans.insert(event.hash(), event); self.reorganize().await; } pub fn is_orphan(&self, event: &Event) -> bool { !self.event_map.contains_key(&event.previous_event_hash) } pub fn find_leaves(&self) -> Vec { // collect the leaves in the tree let mut leaves = vec![]; for (event_hash, node) in self.event_map.iter() { // check if the node is a leaf if node.children.is_empty() { leaves.push(*event_hash); } } leaves } pub fn get_event(&self, event: &EventId) -> Option> { self.event_map.get(event).map(|en| en.event.clone()) } pub fn get_offspring(&self, event: &EventId) -> Vec> { let mut offspring = vec![]; let mut event = *event; let head = self.find_head(); loop { if event == head { break } if let Some(ev) = self.event_map.get(&event) { for child in ev.children.iter() { let child = self.event_map.get(child).unwrap(); offspring.push(child.event.clone()); event = child.event.hash(); } } else { break } } offspring } async fn reorganize(&mut self) { for (_, orphan) in std::mem::take(&mut self.orphans) { // if self.is_orphan(&orphan) { // // TODO should we remove orphan if it's too old // continue // } let prev_event = orphan.previous_event_hash; let node = EventNode { parent: Some(prev_event), event: orphan.clone(), children: Vec::new() }; let node_hash = node.event.hash(); let parent = match self.event_map.get_mut(&prev_event) { Some(parent) => parent, None => { error!("No parent found, Orphan is not relinked"); self.orphans.insert(orphan.hash(), orphan); continue } }; parent.children.push(node_hash); self.event_map.insert(node_hash, node.clone()); self.events_queue.dispatch(&node.event).await.ok(); // clean up the tree from old eventnodes self.prune_chains(); } } fn prune_chains(&mut self) { let head = self.find_head(); let leaves = self.find_leaves(); // Reject events which attach to chains too low in the chain // At some point we ignore all events from old branches for leaf in leaves { // skip the head event if leaf == head { continue } let depth = self.diff_depth(leaf, head); if depth > MAX_DEPTH { self.remove_node(leaf); } } } fn remove_node(&mut self, mut event_id: EventId) { loop { if !self.event_map.contains_key(&event_id) { break } if event_id == self.current_root { break } let node = self.event_map.get(&event_id).unwrap().clone(); self.event_map.remove(&event_id); let parent = self.event_map.get_mut(&node.parent.unwrap()).unwrap(); if parent.children.is_empty() { event_id = parent.event.hash(); continue } let index = parent.children.iter().position(|&n| n == event_id).unwrap(); parent.children.remove(index); event_id = parent.event.hash(); } } // find_head // -> recursively call itself // -> + 1 for every recursion, return self if no children // -> select max from returned values // Gets the lead node with the maximal number of events counting from root fn find_head(&self) -> EventId { self.find_longest_chain(&self.current_root, 0).0 } fn find_longest_chain(&self, parent_node: &EventId, i: u32) -> (EventId, u32) { let children = &self.event_map.get(parent_node).unwrap().children; if children.is_empty() { return (*parent_node, i) } let mut current_max = 0; let mut current_node = None; for node in children.iter() { let (grandchild_node, grandchild_i) = self.find_longest_chain(node, i + 1); match &grandchild_i.cmp(¤t_max) { Ordering::Greater => { current_max = grandchild_i; current_node = Some(grandchild_node); } Ordering::Equal => { // Break ties using the timestamp let grandchild_node_timestamp = self.event_map.get(&grandchild_node).unwrap().event.timestamp; let current_node_timestamp = self.event_map.get(¤t_node.unwrap()).unwrap().event.timestamp; if grandchild_node_timestamp > current_node_timestamp { current_max = grandchild_i; current_node = Some(grandchild_node); } } Ordering::Less => { // Left a todo here, not sure if it should be handled continue } } } assert_ne!(current_max, 0); (current_node.expect("internal logic error"), current_max) } fn find_depth(&self, mut node: EventId, ancestor_id: &EventId) -> u32 { let mut depth = 0; while &node != ancestor_id { depth += 1; if let Some(parent) = self.event_map.get(&node).unwrap().parent { node = parent } else { break } } depth } // Find common ancestor between two events fn find_ancestor(&self, mut node_a: EventId, node_b: EventId) -> EventId { // node_a is a child of node_b let is_child = node_b == self.event_map.get(&node_a).unwrap().parent.unwrap(); if is_child { return node_b } loop { let node_a_parent = self.event_map.get(&node_a).unwrap().parent.unwrap(); node_a = node_a_parent; if node_a == self.current_root { return self.current_root } if self.event_map.get(&node_a).unwrap().children.len() > 1 { let offsprings = self .get_offspring(&node_a) .iter() .map(|event| event.hash()) .collect::>(); if offsprings.contains(&node_b) { return node_a } } } } // Find the length between two events fn diff_depth(&self, node_a: EventId, node_b: EventId) -> u32 { let ancestor = self.find_ancestor(node_a, node_b); let node_a_depth = self.find_depth(node_a, &ancestor); let node_b_depth = self.find_depth(node_b, &ancestor); (node_b_depth + 1).abs_diff(node_a_depth) } fn _debug(&self) { for (event_id, event_node) in &self.event_map { let depth = self.find_depth(*event_id, &self.current_root); println!("{}: {:?} [depth={}]", event_id, event_node.event, depth); } println!("root: {}", self.current_root); println!("head: {}", self.find_head()); } } #[cfg(test)] mod tests { use std::{ fs::{create_dir_all, remove_dir_all}, path::PathBuf, }; use super::*; use crate::{event_graph::events_queue::EventsQueue, util::async_util::sleep, Result}; #[derive(SerialEncodable, SerialDecodable, Clone, Debug)] pub struct PrivMsgEvent { pub nick: String, pub msg: String, pub target: String, } impl std::string::ToString for PrivMsgEvent { fn to_string(&self) -> String { format!(":{}!anon@dark.fi PRIVMSG {} :{}\r\n", self.nick, self.target, self.msg) } } impl EventMsg for PrivMsgEvent { fn new() -> Self { Self { nick: "root".to_string(), msg: "Let there be dark".to_string(), target: "root".to_string(), } } } fn create_message(previous_event_hash: EventId, timestamp: Timestamp) -> Event { Event { previous_event_hash, action: PrivMsgEvent::new(), timestamp } } #[async_std::test] async fn test_remove_old_events() { let events_queue = EventsQueue::new(); let mut model = Model::new(events_queue); let root_id = model.current_root; // event_node 1 // Fill this node with 10 events // These are considered old events from 10 days ago let mut event_node_1_ids = vec![]; let mut id1 = root_id; let timestamp = Timestamp::current_time().0 - 864000; // 864000 is 10 days in seconds for i in 0..10 { let node = create_message(id1, Timestamp(timestamp + i)); id1 = node.hash(); model.add(node).await; event_node_1_ids.push(id1); } sleep(1).await; // event_node 2 // Fill this node with 10 events // These are considered new events at current time let timestamp = Timestamp::current_time().0; for i in 0..150 { let node = create_message(id1, Timestamp(timestamp + i)); id1 = node.hash(); model.add(node).await; } sleep(1).await; // every event older than one week gets removed let ts = Timestamp::current_time().0 - 604800; // one week in seconds let _ = model.remove_old_events(Timestamp(ts)); // ensure the 10 events from event_node 1 are not in the tree anymore for event in event_node_1_ids { assert!(!model.event_map.contains_key(&event)); } // event_node 2 events (150) + root event = 151 events assert_eq!(model.event_map.len(), 151_usize); } #[async_std::test] async fn test_prune_chains() { let events_queue = EventsQueue::new(); let mut model = Model::new(events_queue); let root_id = model.current_root; // event_node 1 // Fill this node with 10 events let mut event_node_1_ids = vec![]; let mut id1 = root_id; for _ in 0..10 { let node = create_message(id1, Timestamp::current_time()); id1 = node.hash(); model.add(node).await; event_node_1_ids.push(id1); } sleep(1).await; // event_node 2 // Start from the root_id and fill the node with (MAX_DEPTH + 10) events. // All the events from event_node_1 should get removed from the tree let mut id2 = root_id; for _ in 0..(MAX_DEPTH + 10) { let node = create_message(id2, Timestamp::current_time()); id2 = node.hash(); model.add(node).await; } assert_eq!(model.find_head(), id2); // Ensure events from node 1 are removed in favor of node 2's longer chain for id in event_node_1_ids { assert!(!model.event_map.contains_key(&id)); } // node1: (10 leaves) + node2: (MAX_DEPTH + 10) events + root event = (MAX_DEPTH + 11) // these ^^^^^^^^^^^ are pruned assert_eq!(model.event_map.len(), (MAX_DEPTH + 11) as usize); } #[async_std::test] async fn test_diff_depth() { let events_queue = EventsQueue::new(); let mut model = Model::new(events_queue); let root_id = model.current_root; // event_node 1 // Fill this node with (MAX_DEPTH / 2) events let mut id1 = root_id; for _ in 0..(MAX_DEPTH / 2) { let node = create_message(id1, Timestamp::current_time()); id1 = node.hash(); model.add(node).await; } sleep(1).await; // event_node 2 // Start from the root_id and fill the node with (MAX_DEPTH + 10) events // all the events must be added since the depth between id1 // and the last head is less than MAX_DEPTH let mut id2 = root_id; for _ in 0..(MAX_DEPTH + 10) { let node = create_message(id2, Timestamp::current_time()); id2 = node.hash(); model.add(node).await; } assert_eq!(model.find_head(), id2); sleep(1).await; // event_node 3 // This will start as new chain, but no events will be added // since the last event's depth is MAX_DEPTH + 10 let mut id3 = root_id; for _ in 0..30 { let node = create_message(id3, Timestamp::current_time()); id3 = node.hash(); model.add(node).await; // ensure events are not added assert!(!model.event_map.contains_key(&id3)); } sleep(1).await; assert_eq!(model.find_head(), id2); // Add more events to the event_node 1 // At the end this chain must overtake the event_node 2 for _ in (MAX_DEPTH / 2)..(MAX_DEPTH + 15) { let node = create_message(id1, Timestamp::current_time()); id1 = node.hash(); model.add(node).await; } assert_eq!(model.find_head(), id1); } #[async_std::test] async fn save_load_model() -> Result<()> { // Setup directories let path = "/tmp/test_model"; remove_dir_all(path).ok(); let path = PathBuf::from(path); create_dir_all(&path)?; // First model let events_queue = EventsQueue::::new(); let mut model1 = Model::new(events_queue); let root_id = model1.current_root; // Create an event let event = create_message(root_id, Timestamp::current_time()); // Add event to first model model1.add(event).await; // Save first model model1.save_tree(&path)?; // Second model let events_queue = EventsQueue::::new(); let mut model2 = Model::new(events_queue); // Load into second model model2.load_tree(&path)?; // Test equality let res = model1.event_map.len() == model2.event_map.len() && model1.event_map.keys().all(|k| model2.event_map.contains_key(k)); assert!(res); remove_dir_all(path).ok(); Ok(()) } #[test] fn test_event_hash() { let events_queue = EventsQueue::::new(); let model = Model::new(events_queue); let root_id = model.current_root; let event = create_message(root_id, Timestamp::current_time()); let event2 = event.clone(); let event_hash = event.hash(); let event2_hash = event2.hash(); assert_eq!(event2_hash, event_hash); } }