bin/darkwikid: WIP implement rpc commands

Cargo.lock

@@ -1349,6 +1349,7 @@ dependencies = [
  "async-std",
  "async-trait",
  "chrono",
+ "colored",
  "crypto_box",
  "ctrlc-async",
  "darkfi",

bin/darkwiki/src/main.rs

@@ -26,6 +26,16 @@ struct Args {
     endpoint: Url,
 }
 
+fn print_patches(value: &Vec<serde_json::Value>) {
+    for res in value {
+        let res = res.as_array().unwrap();
+        let (title, changes) = (res[0].as_str().unwrap(), res[1].as_str().unwrap());
+        println!("FILE: {}", title);
+        println!("{}", changes);
+        println!("----------------------------------");
+    }
+}
+
 #[async_std::main]
 async fn main() -> Result<()> {
     let args = Args::from_args();
@@ -44,6 +54,39 @@ async fn main() -> Result<()> {
         JsonRequest::new("update", json!([]))
     };
 
-    rpc_client.request(req).await?;
+    let result = rpc_client.request(req).await?;
+
+    if !args.log {
+        let result = result.as_array().unwrap();
+        let local_patches = result[0].as_array().unwrap();
+        let sync_patches = result[1].as_array().unwrap();
+        let merge_patches = result[2].as_array().unwrap();
+
+        if !local_patches.is_empty() {
+            println!("");
+            println!("PUBLISH LOCAL PATCHES:");
+            println!("");
+            print_patches(local_patches);
+        }
+
+        if !sync_patches.is_empty() {
+            println!("");
+            println!("RECEIVED PATCHES:");
+            println!("");
+            print_patches(sync_patches);
+        }
+
+        if !merge_patches.is_empty() {
+            println!("");
+            println!("MERGE:");
+            println!("");
+            print_patches(merge_patches);
+        }
+    }
+
+    if args.log {
+        todo!("TODO");
+    }
+
     rpc_client.close().await
 }

bin/darkwikid/Cargo.toml

@@ -34,6 +34,7 @@ thiserror = "1.0.32"
 ctrlc-async = {version= "3.2.2", default-features = false, features = ["async-std", "termination"]}
 url = "2.2.2"
 fxhash = "0.2.1"
+colored = "2.0.0"
 
 # Encoding and parsing
 serde = {version = "1.0.142", features = ["derive"]}

bin/darkwikid/src/jsonrpc.rs

@@ -12,7 +12,8 @@ use darkfi::{
 };
 
 pub struct JsonRpcInterface {
-    update_notifier: async_channel::Sender<()>,
+    sender: async_channel::Sender<String>,
+    receiver: async_channel::Receiver<Vec<Vec<(String, String)>>>,
 }
 
 #[async_trait]
@@ -26,6 +27,8 @@ impl RequestHandler for JsonRpcInterface {
 
         let rep = match req.method.as_str() {
             Some("update") => self.update(req.id, params).await,
+            Some("dry_run") => self.dry_run(req.id, params).await,
+            Some("log") => self.log(req.id, params).await,
             Some(_) | None => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         };
 
@@ -34,22 +37,58 @@ impl RequestHandler for JsonRpcInterface {
 }
 
 impl JsonRpcInterface {
-    pub fn new(update_notifier: async_channel::Sender<()>) -> Self {
-        Self { update_notifier }
+    pub fn new(
+        sender: async_channel::Sender<String>,
+        receiver: async_channel::Receiver<Vec<Vec<(String, String)>>>,
+    ) -> Self {
+        Self { sender, receiver }
     }
 
     // RPCAPI:
     // Update files in ~/darkwiki
     // --> {"jsonrpc": "2.0", "method": "update", "params": [], "id": 1}
-    // <-- {"jsonrpc": "2.0", "result": true, "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": [[(String, String)]], "id": 1}
     async fn update(&self, id: Value, _params: &[Value]) -> JsonResult {
-        let res = self.update_notifier.send(()).await.map_err(Error::from);
+        let res = self.sender.send("update".into()).await.map_err(Error::from);
 
         if let Err(e) = res {
             error!("Failed to update: {}", e);
             return JsonError::new(ErrorCode::InternalError, None, id).into()
         }
 
-        JsonResponse::new(json!(true), id).into()
+        let response = self.receiver.recv().await.unwrap();
+        JsonResponse::new(json!(response), id).into()
+    }
+
+    // RPCAPI:
+    // Update files in darkwiki (dry_run)
+    // --> {"jsonrpc": "2.0", "method": "dry_run", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": [[(String, String)]], "id": 1}
+    async fn dry_run(&self, id: Value, _params: &[Value]) -> JsonResult {
+        let res = self.sender.send("dry_run".into()).await.map_err(Error::from);
+
+        if let Err(e) = res {
+            error!("Failed to update(dry run): {}", e);
+            return JsonError::new(ErrorCode::InternalError, None, id).into()
+        }
+
+        let response = self.receiver.recv().await.unwrap();
+        JsonResponse::new(json!(response), id).into()
+    }
+
+    // RPCAPI:
+    // Show all patches
+    // --> {"jsonrpc": "2.0", "method": "log", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": [[(String, String)]], "id": 1}
+    async fn log(&self, id: Value, _params: &[Value]) -> JsonResult {
+        let res = self.sender.send("log".into()).await.map_err(Error::from);
+
+        if let Err(e) = res {
+            error!("Failed to show all patches: {}", e);
+            return JsonError::new(ErrorCode::InternalError, None, id).into()
+        }
+
+        let response = self.receiver.recv().await.unwrap();
+        JsonResponse::new(json!(response), id).into()
     }
 }

bin/darkwikid/src/main.rs

@@ -38,6 +38,8 @@ use error::DarkWikiResult;
 use jsonrpc::JsonRpcInterface;
 use patch::{OpMethod, Patch};
 
+type Patches = (Vec<Patch>, Vec<Patch>, Vec<Patch>, Vec<Patch>);
+
 pub const CONFIG_FILE: &str = "darkwiki.toml";
 pub const CONFIG_FILE_CONTENTS: &str = include_str!("../darkwiki.toml");
 
@@ -146,8 +148,11 @@ fn title_to_id(title: &str) -> String {
     hex::encode(hasher.finalize())
 }
 
-fn on_receive_update(settings: &DarkWikiSettings) -> DarkWikiResult<Vec<Patch>> {
+fn on_receive_update(settings: &DarkWikiSettings, dry: bool) -> DarkWikiResult<Patches> {
     let mut patches: Vec<Patch> = vec![];
+    let mut local_patches: Vec<Patch> = vec![];
+    let mut sync_patches: Vec<Patch> = vec![];
+    let mut merge_patches: Vec<Patch> = vec![];
 
     let local_path = settings.datastore_path.join("local");
     let sync_path = settings.datastore_path.join("sync");
@@ -187,12 +192,19 @@ fn on_receive_update(settings: &DarkWikiSettings) -> DarkWikiResult<Vec<Patch>>
 
             new_patch.base = local_patch.to_string();
 
+            local_patches.push(new_patch.clone());
+
             // check if the same doc has received patch from the network
             if let Ok(sync_patch) = load_json_file::<Patch>(&sync_path.join(&doc_id)) {
                 if sync_patch.timestamp != local_patch.timestamp {
+                    sync_patches.push(sync_patch.clone());
+
                     let sync_patch_t = new_patch.transform(&sync_patch);
                     new_patch = new_patch.merge(&sync_patch_t);
-                    save_file(&docs_path.join(doc_title), &new_patch.to_string())?;
+                    if !dry {
+                        save_file(&docs_path.join(doc_title), &new_patch.to_string())?;
+                    }
+                    merge_patches.push(new_patch.clone());
                 }
             }
 
@@ -200,11 +212,14 @@ fn on_receive_update(settings: &DarkWikiSettings) -> DarkWikiResult<Vec<Patch>>
             b_patch.base = "".to_string();
         } else {
             new_patch.base = edit.to_string();
+            local_patches.push(new_patch.clone());
             b_patch = new_patch.clone();
         };
 
-        save_json_file(&local_path.join(&doc_id), &new_patch)?;
-        save_json_file(&sync_path.join(doc_id), &new_patch)?;
+        if !dry {
+            save_json_file(&local_path.join(&doc_id), &new_patch)?;
+            save_json_file(&sync_path.join(doc_id), &new_patch)?;
+        }
         patches.push(b_patch);
     }
 
@@ -223,26 +238,56 @@ fn on_receive_update(settings: &DarkWikiSettings) -> DarkWikiResult<Vec<Patch>>
             }
         }
 
-        save_file(&docs_path.join(&sync_patch.title), &sync_patch.to_string())?;
-        save_json_file(&local_path.join(file_id), &sync_patch)?;
+        if !dry {
+            save_file(&docs_path.join(&sync_patch.title), &sync_patch.to_string())?;
+            save_json_file(&local_path.join(file_id), &sync_patch)?;
+        }
+
+        sync_patches.push(sync_patch);
     }
 
-    Ok(patches)
+    Ok((patches, local_patches, sync_patches, merge_patches))
 }
 
 async fn start(
-    update_notifier_rv: async_channel::Receiver<()>,
+    rpc_rv: async_channel::Receiver<String>,
+    notify_sx: async_channel::Sender<Vec<Vec<(String, String)>>>,
     raft_sender: async_channel::Sender<Patch>,
     raft_receiver: async_channel::Receiver<Patch>,
     settings: DarkWikiSettings,
 ) -> DarkWikiResult<()> {
     loop {
         select! {
-            _ = update_notifier_rv.recv().fuse() => {
-                let patches = on_receive_update(&settings)?;
-                for patch in patches {
-                    info!("Send a patch to Raft {:?}", patch);
-                    raft_sender.send(patch).await.map_err(Error::from)?;
+            command = rpc_rv.recv().fuse() => {
+                let command = command.unwrap();
+                match command.as_str() {
+                    "update" | "dry_run" => {
+                        let dry = command.as_str() == "dry_run";
+                        let (patches, local, sync, merge) = on_receive_update(&settings, dry)?;
+
+                        if !dry {
+                            for patch in patches {
+                                info!("Send a patch to Raft {:?}", patch);
+                                raft_sender.send(patch.clone()).await.map_err(Error::from)?;
+                            }
+                        }
+
+                        let local: Vec<(String, String)> =
+                            local.iter().map(|p| (p.title.to_owned(), p.colorize())).collect();
+
+                        let sync: Vec<(String, String)> =
+                            sync.iter().map(|p| (p.title.to_owned(), p.colorize())).collect();
+
+                        let merge: Vec<(String, String)> =
+                            merge.iter().map(|p| (p.title.to_owned(), p.colorize())).collect();
+
+                        notify_sx.send(vec![local, sync, merge]).await.map_err(Error::from)?;
+                    }
+                    "log" => {
+                        // TODO
+                        notify_sx.send(vec![]).await.map_err(Error::from)?;
+                    }
+                    _ => {}
                 }
             }
             patch = raft_receiver.recv().fuse() => {
@@ -264,12 +309,13 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     create_dir_all(datastore_path.join("local"))?;
     create_dir_all(datastore_path.join("sync"))?;
 
-    let (update_notifier_sx, update_notifier_rv) = async_channel::unbounded::<()>();
+    let (rpc_sx, rpc_rv) = async_channel::unbounded::<String>();
+    let (notify_sx, notify_rv) = async_channel::unbounded::<Vec<Vec<(String, String)>>>();
 
     //
     // RPC
     //
-    let rpc_interface = Arc::new(JsonRpcInterface::new(update_notifier_sx));
+    let rpc_interface = Arc::new(JsonRpcInterface::new(rpc_sx, notify_rv));
     executor.spawn(listen_and_serve(settings.rpc_listen.clone(), rpc_interface)).detach();
 
     //
@@ -314,7 +360,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     //
     let darkwiki_settings = DarkWikiSettings { author: settings.author, datastore_path, docs_path };
     executor
-        .spawn(start(update_notifier_rv, raft.sender(), raft.receiver(), darkwiki_settings))
+        .spawn(start(rpc_rv, notify_sx, raft.sender(), raft.receiver(), darkwiki_settings))
         .detach();
 
     //

bin/darkwikid/src/patch.rs

@@ -1,5 +1,6 @@
 use std::{cmp::Ordering, io};
 
+use colored::Colorize;
 use serde::{Deserialize, Serialize};
 
 use darkfi::util::{
@@ -31,39 +32,36 @@ pub struct Patch {
 
 impl std::string::ToString for Patch {
     fn to_string(&self) -> String {
-        let mut st = vec![];
-        let mut index: usize = 0;
+        if self.ops.0.is_empty() {
+            return self.base.clone()
+        }
 
+        let mut st = vec![];
         st.extend(str_to_chars(&self.base));
+        let st = &mut st.iter();
+
+        let mut new_st: Vec<&str> = vec![];
+
         for op in self.ops.0.iter() {
             match op {
                 OpMethod::Retain(n) => {
-                    index += *n as usize;
+                    for c in st.take(*n as usize) {
+                        new_st.push(c);
+                    }
                 }
                 OpMethod::Delete(n) => {
-                    if (index + (*n as usize)) > st.len() {
-                        if index < st.len() {
-                            st.drain(index..st.len());
-                        }
-                    } else {
-                        st.drain(index..(index + *n as usize));
+                    for _ in 0..*n {
+                        st.next();
                     }
                 }
                 OpMethod::Insert(insert) => {
                     let chars = str_to_chars(insert);
-                    for c in chars {
-                        if index > st.len() {
-                            st.push(c);
-                        } else {
-                            st.insert(index, c);
-                            index += 1;
-                        }
-                    }
+                    new_st.extend(chars);
                 }
             }
         }
 
-        st.join("")
+        new_st.join("")
     }
 }
 
@@ -157,14 +155,19 @@ impl Patch {
         let mut op1 = ops1.next();
         let mut op2 = ops2.next();
         loop {
-            if op2.is_none() {
-                break
-            }
-
-            if op1.is_none() {
-                new_patch.add_op(op2.as_ref().unwrap());
-                op2 = ops2.next();
-                continue
+            match (&op1, &op2) {
+                (None, None) => break,
+                (None, Some(op)) => {
+                    new_patch.add_op(op);
+                    op2 = ops2.next();
+                    continue
+                }
+                (Some(op), None) => {
+                    new_patch.add_op(op);
+                    op1 = ops1.next();
+                    continue
+                }
+                _ => {}
             }
 
             match (op1.as_ref().unwrap(), op2.as_ref().unwrap()) {
@@ -257,14 +260,19 @@ impl Patch {
         let mut op2 = ops2.next();
 
         loop {
-            if op2.is_none() {
-                break
-            }
-
-            if op1.is_none() {
-                new_patch.add_op(op2.as_ref().unwrap());
-                op2 = ops2.next();
-                continue
+            match (&op1, &op2) {
+                (None, None) => break,
+                (None, Some(op)) => {
+                    new_patch.add_op(op);
+                    op2 = ops2.next();
+                    continue
+                }
+                (Some(op), None) => {
+                    new_patch.add_op(op);
+                    op1 = ops1.next();
+                    continue
+                }
+                _ => {}
             }
 
             match (op1.as_ref().unwrap(), op2.as_ref().unwrap()) {
@@ -356,6 +364,44 @@ impl Patch {
 
         new_patch
     }
+
+    pub fn colorize(&self) -> String {
+        if self.ops.0.is_empty() {
+            return format!("{}", self.base.green())
+        }
+
+        let mut st = vec![];
+        st.extend(str_to_chars(&self.base));
+        let st = &mut st.iter();
+
+        let mut colorized_str: Vec<String> = vec![];
+
+        for op in self.ops.0.iter() {
+            match op {
+                OpMethod::Retain(n) => {
+                    for c in st.take(*n as usize) {
+                        colorized_str.push(c.to_string());
+                    }
+                }
+                OpMethod::Delete(n) => {
+                    let mut deleted_part = vec![];
+                    for _ in 0..*n {
+                        let s = st.next();
+                        if let Some(s) = s {
+                            deleted_part.push(s.to_string());
+                        }
+                    }
+                    colorized_str.push(format!("{}", deleted_part.join("").red()));
+                }
+                OpMethod::Insert(insert) => {
+                    let chars = str_to_chars(insert);
+                    colorized_str.push(format!("{}", chars.join("").green()));
+                }
+            }
+        }
+
+        colorized_str.join("")
+    }
 }
 
 impl Decodable for OpMethod {

src/raft/consensus.rs

@@ -167,14 +167,12 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             }
 
             // send pending messages
-            if !self.pending_msgs.is_empty() {
-                if self.role != Role::Candidate {
-                    let pending_msgs = self.pending_msgs.clone();
-                    for m in &pending_msgs {
-                        result = self.broadcast_msg(m, None).await;
-                    }
-                    self.pending_msgs = vec![];
+            if !self.pending_msgs.is_empty() && self.role != Role::Candidate {
+                let pending_msgs = self.pending_msgs.clone();
+                for m in &pending_msgs {
+                    result = self.broadcast_msg(m, None).await;
                 }
+                self.pending_msgs = vec![];
             }
 
             match result {