wallet: switch from evgrd to a true p2p backend

bin/darkwallet/Cargo.lock

@@ -2862,8 +2862,8 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
 [[package]]
 name = "miniquad"
-version = "0.4.6"
-source = "git+https://github.com/not-fl3/miniquad#5a9a76b702a9d3fa3e937206879b2cb0c7827ef3"
+version = "0.4.7"
+source = "git+https://github.com/narodnik/miniquad#fc1fc9bbf11cd14137c5b37ed1a8590383cb20b1"
 dependencies = [
  "libc",
  "ndk-sys",

bin/darkwallet/Cargo.toml

@@ -13,7 +13,7 @@ repository = "https://codeberg.org/darkrenaissance/darkfi"
 #path = "bin/drawsim.rs"
 
 [dependencies]
-miniquad = { git = "https://github.com/not-fl3/miniquad" }
+miniquad = { git = "https://github.com/narodnik/miniquad" }
 
 # Currently latest version links to freetype-sys 0.19 but we use 0.21
 #harfbuzz-sys = "0.6.1"
@@ -67,6 +67,7 @@ android_logger = "0.13.3"
 # Required by Arti: tor-dirmgr
 rusqlite = {version = "0.32.1", features = ["bundled"]}
 tor-dirmgr = {version="0.23.0", features=["static"]}
+#android-fileopen = {path = "./android-fileopen/"}
 
 [package.metadata.android.activity_attributes]
 "android:exported" = "true"

bin/darkwallet/src/app/mod.rs

@@ -29,7 +29,7 @@ use std::{
 };
 
 use crate::{
-    darkirc2::LocalDarkIRCPtr,
+    darkirc::DarkIrcBackendPtr,
     error::Error,
     expr::Op,
     gfx::{GraphicsEventPublisherPtr, RenderApiPtr, Vertex},
@@ -121,7 +121,7 @@ pub struct App {
     pub render_api: RenderApiPtr,
     pub event_pub: GraphicsEventPublisherPtr,
     pub text_shaper: TextShaperPtr,
-    pub darkirc_evgr: SyncMutex<Option<LocalDarkIRCPtr>>,
+    pub darkirc_evgr: SyncMutex<Option<DarkIrcBackendPtr>>,
     pub tasks: SyncMutex<Vec<Task<()>>>,
     pub ex: ExecutorPtr,
 }

bin/darkwallet/src/darkirc.rs

@@ -16,7 +16,10 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::{Arc, Mutex as SyncMutex};
+use std::{
+    sync::{Arc, Mutex as SyncMutex},
+    time::UNIX_EPOCH,
+};
 
 use darkfi::{
     event_graph::{
@@ -29,16 +32,28 @@ use darkfi::{
     Error,
 };
 use darkfi_serial::{
-    async_trait, deserialize_async, serialize_async, Encodable, SerialDecodable, SerialEncodable,
+    async_trait, deserialize_async, serialize_async, AsyncEncodable, Encodable, SerialDecodable,
+    SerialEncodable,
 };
 use sled_overlay::sled;
 
-use crate::{scene::SceneGraphPtr2, ExecutorPtr};
+use crate::{
+    prop::{PropertyBool, PropertyFloat32, PropertyStr, Role},
+    scene::{SceneNodePtr, Slot},
+    ui::chatview::MessageId,
+    ExecutorPtr,
+};
 
 #[cfg(target_os = "android")]
-const EVGRDB_PATH: &str = "/data/data/darkfi.darkwallet/evgrdb/";
-#[cfg(target_os = "linux")]
-const EVGRDB_PATH: &str = "evgrdb";
+const EVGRDB_PATH: &str = "/data/data/darkfi.darkwallet/evgr/";
+#[cfg(not(target_os = "android"))]
+const EVGRDB_PATH: &str = "~/.local/darkfi/darkwallet/evgr/";
+
+const CHANNEL: &str = "#random";
+
+/// Due to drift between different machine's clocks, if the message timestamp is recent
+/// then we will just correct it to the current time so messages appear sequential in the UI.
+const RECENT_TIME_DIST: u64 = 10_000;
 
 #[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 pub struct Privmsg {
@@ -47,63 +62,57 @@ pub struct Privmsg {
     pub msg: String,
 }
 
-async fn relay_darkirc_events(sg: SceneGraphPtr2, ev_sub: Subscription<event_graph::Event>) {
-    loop {
-        let ev = ev_sub.receive().await;
-
-        // Try to deserialize the `Event`'s content into a `Privmsg`
-        let privmsg: Privmsg = match deserialize_async(ev.content()).await {
-            Ok(v) => v,
-            Err(e) => {
-                error!("[IRC CLIENT] Failed deserializing incoming Privmsg event: {}", e);
-                continue
-            }
-        };
-
-        if privmsg.channel != "#random" {
-            continue
-        }
-
-        info!(target: "darkirc", "ev_id={:?}", ev.id());
-        info!(target: "darkirc", "ev: {:?}", ev);
-        info!(target: "darkirc", "privmsg: {:?}", privmsg);
-        info!(target: "darkirc", "");
-
-        let response_fn = Box::new(|_| {});
+impl Privmsg {
+    pub fn new(channel: String, nick: String, msg: String) -> Self {
+        Self { channel, nick, msg }
+    }
 
-        let mut arg_data = vec![];
-        ev.timestamp.encode(&mut arg_data).unwrap();
-        ev.id().as_bytes().encode(&mut arg_data).unwrap();
-        privmsg.nick.encode(&mut arg_data).unwrap();
-        privmsg.msg.encode(&mut arg_data).unwrap();
-
-        let mut sg = sg.lock().await;
-        let chatview_node = sg.lookup_node_mut("/window/view/chatty").unwrap();
-        chatview_node.call_method("insert_line", arg_data, response_fn).unwrap();
-        drop(sg);
+    pub fn msg_id(&self, timest: u64) -> MessageId {
+        let mut hasher = blake3::Hasher::new();
+        timest.encode(&mut hasher).unwrap();
+        self.channel.encode(&mut hasher).unwrap();
+        self.nick.encode(&mut hasher).unwrap();
+        self.msg.encode(&mut hasher).unwrap();
+        MessageId(hasher.finalize().into())
     }
 }
 
 pub type DarkIrcBackendPtr = Arc<DarkIrcBackend>;
 
-struct DarkIrcData {
+pub struct DarkIrcBackend {
+    ex: ExecutorPtr,
     p2p: P2pPtr,
     event_graph: EventGraphPtr,
-    #[allow(dead_code)]
-    ev_task: smol::Task<()>,
+    tasks: SyncMutex<Vec<smol::Task<()>>>,
     db: sled::Db,
-}
 
-pub struct DarkIrcBackend(SyncMutex<Option<DarkIrcData>>);
+    chatview_node: SceneNodePtr,
+    sendbtn_node: SceneNodePtr,
+    editbox_node: SceneNodePtr,
+    editbox_text: PropertyStr,
+    chatview_scroll: PropertyFloat32,
+    upgrade_popup_is_visible: PropertyBool,
+
+    seen_msgs: SyncMutex<Vec<MessageId>>,
+}
 
 impl DarkIrcBackend {
-    pub fn new() -> Arc<Self> {
-        Arc::new(Self(SyncMutex::new(None)))
-    }
+    pub async fn new(sg_root: SceneNodePtr, ex: ExecutorPtr) -> darkfi::Result<Arc<Self>> {
+        let chatview_node = sg_root.clone().lookup_node("/window/view/chatty").unwrap();
+        let sendbtn_node = sg_root.clone().lookup_node("/window/view/send_btn").unwrap();
+
+        let editbox_node = sg_root.clone().lookup_node("/window/view/editz").unwrap();
+        let editbox_text = PropertyStr::wrap(&editbox_node, Role::App, "text", 0).unwrap();
+
+        let chatview_scroll =
+            PropertyFloat32::wrap(&chatview_node, Role::Internal, "scroll", 0).unwrap();
+
+        let upgrade_popup_node = sg_root.clone().lookup_node("/window/view/upgrade_popup").unwrap();
+        let upgrade_popup_is_visible =
+            PropertyBool::wrap(&upgrade_popup_node, Role::App, "is_visible", 0).unwrap();
 
-    pub async fn start(&self, sg: SceneGraphPtr2, ex: ExecutorPtr) -> darkfi::Result<()> {
         info!(target: "darkirc", "Starting DarkIRC backend");
-        let sled_db = sled::open(EVGRDB_PATH)?;
+        let db = sled::open(EVGRDB_PATH)?;
 
         let mut p2p_settings: NetSettings = Default::default();
         p2p_settings.app_version = semver::Version::parse("0.5.0").unwrap();
@@ -113,7 +122,7 @@ impl DarkIrcBackend {
 
         let event_graph = EventGraph::new(
             p2p.clone(),
-            sled_db.clone(),
+            db.clone(),
             std::path::PathBuf::new(),
             false,
             "darkirc_dag",
@@ -122,11 +131,30 @@ impl DarkIrcBackend {
         )
         .await?;
 
+        Ok(Arc::new(Self {
+            ex,
+            p2p,
+            event_graph,
+            tasks: SyncMutex::new(vec![]),
+            db,
+
+            chatview_node,
+            sendbtn_node,
+            editbox_node,
+            editbox_text,
+            chatview_scroll,
+            upgrade_popup_is_visible,
+
+            seen_msgs: SyncMutex::new(vec![]),
+        }))
+    }
+
+    pub async fn start(self: Arc<Self>, ex: ExecutorPtr) -> darkfi::Result<()> {
         //self.prune_task.lock().unwrap() = Some(event_graph.prune_task.get().unwrap());
 
         info!(target: "darkirc", "Registering EventGraph P2P protocol");
-        let event_graph_ = Arc::clone(&event_graph);
-        let registry = p2p.protocol_registry();
+        let event_graph_ = Arc::clone(&self.event_graph);
+        let registry = self.p2p.protocol_registry();
         registry
             .register(SESSION_DEFAULT, move |channel, _| {
                 let event_graph_ = event_graph_.clone();
@@ -134,11 +162,47 @@ impl DarkIrcBackend {
             })
             .await;
 
-        let ev_sub = event_graph.event_pub.clone().subscribe().await;
-        let ev_task = ex.spawn(relay_darkirc_events(sg, ev_sub));
+        let ev_sub = self.event_graph.event_pub.clone().subscribe().await;
+        let ev_task = self.ex.spawn(self.clone().relay_events(ev_sub));
 
         info!(target: "darkirc", "Starting P2P network");
-        p2p.clone().start().await?;
+        self.p2p.clone().start().await?;
+
+        // Connect the UI send up
+
+        let (slot, recvr) = Slot::new("send_button_clicked");
+        self.sendbtn_node.register("click", slot).unwrap();
+        let me = Arc::downgrade(&self);
+        let send_task = ex.spawn(async move {
+            while let Some(self_) = me.upgrade() {
+                let Ok(_) = recvr.recv().await else {
+                    error!(target: "ui::win", "Button click recvr closed");
+                    break
+                };
+                self_.handle_send().await;
+            }
+        });
+
+        let (slot, recvr) = Slot::new("enter_pressed");
+        self.editbox_node.register("enter_pressed", slot).unwrap();
+        let me = Arc::downgrade(&self);
+        let enter_task = ex.spawn(async move {
+            while let Some(self_) = me.upgrade() {
+                let Ok(_) = recvr.recv().await else {
+                    error!(target: "ui::win", "EditBox enter_pressed recvr closed");
+                    break
+                };
+                self_.handle_send().await;
+            }
+        });
+
+        {
+            let mut tasks = self.tasks.lock().unwrap();
+            assert!(tasks.is_empty());
+            *tasks = vec![send_task, enter_task];
+        }
+
+        // Sync the DAG
 
         info!(target: "darkirc", "Waiting for some P2P connections...");
         sleep(5).await;
@@ -147,12 +211,12 @@ impl DarkIrcBackend {
         let sync_attempts = 4;
         for i in 1..=sync_attempts {
             info!(target: "darkirc", "Syncing event DAG (attempt #{})", i);
-            match event_graph.dag_sync().await {
+            match self.event_graph.dag_sync().await {
                 Ok(()) => break,
                 Err(e) => {
                     if i == sync_attempts {
                         error!("Failed syncing DAG. Exiting.");
-                        p2p.stop().await;
+                        self.p2p.stop().await;
                         return Err(Error::DagSyncFailed)
                     } else {
                         // TODO: Maybe at this point we should prune or something?
@@ -164,28 +228,21 @@ impl DarkIrcBackend {
             }
         }
 
-        *self.0.lock().unwrap() = Some(DarkIrcData { p2p, event_graph, ev_task, db: sled_db });
-
         Ok(())
     }
 
     pub async fn stop(&self) {
         info!(target: "darkirc", "Stopping DarkIRC backend");
-        let self_ = self.0.lock().unwrap();
-        let Some(self_) = &*self_ else {
-            warn!(target: "darkirc", "Backend wasn't started");
-            return
-        };
 
         info!(target: "darkirc", "Stopping P2P network");
-        self_.p2p.stop().await;
+        self.p2p.stop().await;
 
-        info!(target: "darkirc", "Stopping IRC server");
-        let prune_task = self_.event_graph.prune_task.get().unwrap();
+        info!(target: "darkirc", "Stopping event graph prune task");
+        let prune_task = self.event_graph.prune_task.get().unwrap();
         prune_task.stop().await;
 
         info!(target: "darkirc", "Flushing event graph sled database...");
-        let Ok(flushed_bytes) = self_.db.flush_async().await else {
+        let Ok(flushed_bytes) = self.db.flush_async().await else {
             error!(target: "darkirc", "Flushing event graph db failed");
             return
         };
@@ -193,20 +250,88 @@ impl DarkIrcBackend {
         info!(target: "darkirc", "Shut down backend successfully");
     }
 
-    pub async fn send(&self, privmsg: Privmsg) {
-        let (p2p, evgr) = {
-            let self_ = self.0.lock().unwrap();
-            let data = self_.as_ref().expect("backend wasnt started");
+    async fn relay_events(self: Arc<Self>, ev_sub: Subscription<event_graph::Event>) {
+        loop {
+            let ev = ev_sub.receive().await;
 
-            let evgr = data.event_graph.clone();
-            let p2p = data.p2p.clone();
-            (p2p, evgr)
-        };
-        let event = event_graph::Event::new(serialize_async(&privmsg).await, &evgr).await;
+            // Try to deserialize the `Event`'s content into a `Privmsg`
+            let privmsg: Privmsg = match deserialize_async(ev.content()).await {
+                Ok(v) => v,
+                Err(e) => {
+                    error!("[IRC CLIENT] Failed deserializing incoming Privmsg event: {}", e);
+                    continue
+                }
+            };
+
+            if privmsg.channel != CHANNEL {
+                continue
+            }
+
+            info!(target: "darkirc", "ev_id={:?}", ev.id());
+            info!(target: "darkirc", "ev: {:?}", ev);
+            info!(target: "darkirc", "privmsg: {:?}", privmsg);
+            info!(target: "darkirc", "");
+
+            let timest = ev.timestamp;
+            // This is a hack to make messages appear sequentially in the UI
+            let mut adj_timest = timest;
+            let now_timest = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
+            if timest.abs_diff(now_timest) < RECENT_TIME_DIST {
+                debug!(target: "darkirc", "Applied timestamp correction: <{timest}> => <{now_timest}>");
+                adj_timest = now_timest;
+            }
+
+            let msg_id = privmsg.msg_id(timest);
+            {
+                let mut seen = self.seen_msgs.lock().unwrap();
+                if seen.contains(&msg_id) {
+                    warn!(target: "darkirc", "Skipping duplicate seen message: {msg_id}");
+                    continue
+                }
+                seen.push(msg_id.clone());
+            }
+
+            let mut arg_data = vec![];
+            ev.timestamp.encode(&mut arg_data).unwrap();
+            ev.id().as_bytes().encode(&mut arg_data).unwrap();
+            privmsg.nick.encode(&mut arg_data).unwrap();
+            privmsg.msg.encode(&mut arg_data).unwrap();
+
+            self.chatview_node.call_method("insert_line", arg_data).await.unwrap();
+        }
+    }
+
+    async fn handle_send(&self) {
+        // Get text from editbox
+        let text = self.editbox_text.get();
+        if text.is_empty() {
+            return
+        }
+        // Clear editbox
+        self.editbox_text.set("");
+        self.chatview_scroll.set(0.);
+
+        // Send text to channel
+        let timest = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
+        debug!(target: "darkirc", "Sending privmsg: <{timest}> {text}");
+        let msg = Privmsg::new(CHANNEL.to_string(), "anon".to_string(), text);
+
+        let mut arg_data = vec![];
+        timest.encode_async(&mut arg_data).await.unwrap();
+        msg.msg_id(timest).encode_async(&mut arg_data).await.unwrap();
+        msg.nick.encode_async(&mut arg_data).await.unwrap();
+        msg.msg.encode_async(&mut arg_data).await.unwrap();
+
+        self.chatview_node.call_method("insert_unconf_line", arg_data).await.unwrap();
+
+        // Broadcast the msg
+
+        let evgr = self.event_graph.clone();
+        let event = event_graph::Event::new(serialize_async(&msg).await, &evgr).await;
         if let Err(e) = evgr.dag_insert(&[event.clone()]).await {
             error!(target: "darkirc", "Failed inserting new event to DAG: {}", e);
         }
 
-        p2p.broadcast(&EventPut(event)).await;
+        self.p2p.broadcast(&EventPut(event)).await;
     }
 }

bin/darkwallet/src/gfx/linalg.rs

@@ -17,8 +17,8 @@
  */
 
 use async_trait::async_trait;
+use darkfi_serial::{SerialDecodable, SerialEncodable};
 use std::ops::{Add, AddAssign, Div, Mul, Sub, SubAssign};
-use darkfi_serial::{SerialEncodable, SerialDecodable};
 
 #[derive(Clone, Copy, Debug, SerialEncodable, SerialDecodable)]
 pub struct Dimension {

bin/darkwallet/src/gfx/mod.rs

@@ -17,6 +17,7 @@
  */
 
 use darkfi_serial::{async_trait, SerialDecodable, SerialEncodable};
+use futures::AsyncWriteExt;
 use log::debug;
 use miniquad::{
     conf, window, Backend, Bindings, BlendFactor, BlendState, BlendValue, BufferLayout,
@@ -29,7 +30,6 @@ use std::{
     sync::{mpsc, Arc, Mutex as SyncMutex},
     time::{Duration, Instant},
 };
-use futures::AsyncWriteExt;
 
 mod linalg;
 pub use linalg::{Dimension, Point, Rectangle};
@@ -535,7 +535,7 @@ impl Stage {
             buffers: HashMap::new(),
             method_rep,
             event_pub,
-            draw_log: if DEBUG_DRAW_LOG { Some(scr::DrawLog::new()) } else { None }
+            draw_log: if DEBUG_DRAW_LOG { Some(scr::DrawLog::new()) } else { None },
         }
     }
 

bin/darkwallet/src/gfx/scr.rs

@@ -17,8 +17,13 @@
  */
 
 use async_trait::async_trait;
-use darkfi_serial::{SerialEncodable, SerialDecodable, serialize, Encodable, Decodable, deserialize};
-use std::{fs::{OpenOptions, File}, time::Instant};
+use darkfi_serial::{
+    deserialize, serialize, Decodable, Encodable, SerialDecodable, SerialEncodable,
+};
+use std::{
+    fs::{File, OpenOptions},
+    time::Instant,
+};
 
 use super::GraphicsMethod;
 
@@ -27,7 +32,7 @@ const FILENAME: &str = "drawinstrs.dat";
 #[derive(Debug, SerialEncodable, SerialDecodable)]
 struct Instruction {
     timest: u64,
-    method: GraphicsMethod
+    method: GraphicsMethod,
 }
 
 pub struct DrawLog {
@@ -39,17 +44,11 @@ impl DrawLog {
     pub fn new() -> Self {
         let instant = Instant::now();
         let fd = OpenOptions::new().write(true).create(true).open(FILENAME).unwrap();
-        Self {
-            instant,
-            fd,
-        }
+        Self { instant, fd }
     }
 
     pub fn log(&mut self, method: GraphicsMethod) {
-        let instr = Instruction {
-            timest: self.instant.elapsed().as_millis() as u64,
-            method
-        };
+        let instr = Instruction { timest: self.instant.elapsed().as_millis() as u64, method };
         let data = serialize(&instr);
         data.encode(&mut self.fd).unwrap();
     }
@@ -65,4 +64,3 @@ impl DrawLog {
         instrs
     }
 }
-

bin/darkwallet/src/main.rs

@@ -49,7 +49,7 @@ use log::LevelFilter;
 
 mod app;
 mod build_info;
-//mod darkirc;
+mod darkirc;
 mod darkirc2;
 mod error;
 mod expr;
@@ -69,7 +69,7 @@ mod ui;
 mod util;
 
 use crate::{
-    darkirc2::{LocalDarkIRC, LocalDarkIRCPtr},
+    darkirc::{DarkIrcBackend, DarkIrcBackendPtr},
     net::ZeroMQAdapter,
     text::TextShaper,
 };
@@ -82,6 +82,18 @@ fn panic_hook(panic_info: &std::panic::PanicInfo) {
     std::process::exit(1);
 }
 
+/*
+async fn whomain() {
+    use std::sync::Mutex as SyncMutex;
+    let file_data = Arc::new(SyncMutex::new(None));
+    android_fileopen::find_file(file_data.clone());
+
+    //if let Some(ref file_data) = &*file_data.lock().unwrap() {
+    //    info!("content byte length: {}", file_data.len());
+    //}
+}
+*/
+
 fn main() {
     // Exit the application on panic right away
     std::panic::set_hook(Box::new(panic_hook));
@@ -148,7 +160,7 @@ fn main() {
     let ex2 = ex.clone();
     let darkirc_task = ex.spawn(async move {
         cv_started.wait().await;
-        let darkirc_evgr = LocalDarkIRC::new(sg_root.clone(), ex2.clone()).await.unwrap();
+        let darkirc_evgr = DarkIrcBackend::new(sg_root.clone(), ex2.clone()).await.unwrap();
         *app2.darkirc_evgr.lock().unwrap() = Some(darkirc_evgr.clone());
         if let Err(e) = darkirc_evgr.start(ex2).await {
             error!("DarkIRC error: {e}")