Discover and subscribe to documents based on short code

aardvark-app/src/application.rs

@@ -20,13 +20,13 @@
 
 use std::cell::{OnceCell, RefCell};
 
-use aardvark_node::network;
+use aardvark_node::network::{self, FromApp, ToApp};
 use adw::prelude::*;
 use adw::subclass::prelude::*;
+use automerge::PatchAction;
 use gettextrs::gettext;
 use gtk::{gio, glib};
 use tokio::sync::{mpsc, oneshot};
-use automerge::PatchAction;
 
 use crate::config::VERSION;
 use crate::document::Document;
@@ -40,14 +40,13 @@ mod imp {
     pub struct AardvarkApplication {
         pub window: OnceCell<AardvarkWindow>,
         pub document: Document,
-        pub tx: mpsc::Sender<Vec<u8>>,
-        pub rx: RefCell<Option<mpsc::Receiver<Vec<u8>>>>,
+        pub tx: mpsc::Sender<FromApp>,
+        pub rx: RefCell<Option<mpsc::Receiver<ToApp>>>,
         #[allow(dead_code)]
         backend_shutdown: oneshot::Sender<()>,
     }
 
-    impl AardvarkApplication {
-    }
+    impl AardvarkApplication {}
 
     #[glib::object_subclass]
     impl ObjectSubclass for AardvarkApplication {
@@ -143,15 +142,29 @@ impl AardvarkApplication {
 
             {
                 let application = self.clone();
+
+                let tx = self.imp().tx.clone();
+
                 let mut rx = self
                     .imp()
                     .rx
                     .take()
                     .expect("rx should be given at this point");
 
                 glib::spawn_future_local(async move {
-                    while let Some(bytes) = rx.recv().await {
-                        application.ingest_message(bytes);
+                    tx.send(FromApp::CreateDocument)
+                        .await
+                        .expect("app tx channel is open");
+
+                    while let Some(message) = rx.recv().await {
+                        match message {
+                            ToApp::SubscriptionSuccess(text_document) => {
+                                // @TODO: get the short code for this document into the share UI
+                                // component.
+                                println!("new document: {}", text_document.hash())
+                            }
+                            ToApp::MessageReceived(bytes) => application.ingest_message(bytes),
+                        }
                     }
                 });
             }
@@ -182,24 +195,16 @@ impl AardvarkApplication {
 
         // Apply remote changes to our local text CRDT
         if let Err(err) = document.load_incremental(&message) {
-            eprintln!(
-                "failed applying text change from remote peer to automerge document: {err}"
-            );
-            window.add_toast(adw::Toast::new(
-                "The network provided bad data!"
-            ));
+            eprintln!("failed applying text change from remote peer to automerge document: {err}");
+            window.add_toast(adw::Toast::new("The network provided bad data!"));
             return;
         }
 
         // Get latest changes and apply them to our local text buffer
         for patch in document.diff_incremental() {
             match &patch.action {
                 PatchAction::SpliceText { index, value, .. } => {
-                    buffer.splice(
-                        *index as i32,
-                        0,
-                        value.make_string().as_str(),
-                    );
+                    buffer.splice(*index as i32, 0, value.make_string().as_str());
                 }
                 PatchAction::DeleteSeq { index, length } => {
                     buffer.splice(*index as i32, *length as i32, "");
@@ -210,7 +215,9 @@ impl AardvarkApplication {
 
         // Sanity check that the text buffer and CRDT are in the same state
         if buffer.full_text() != document.text() {
-            window.add_toast(adw::Toast::new("The CRDT and the text view have different states!"));
+            window.add_toast(adw::Toast::new(
+                "The CRDT and the text view have different states!",
+            ));
             // if the state diverged, use the CRDT as the source of truth
             buffer.set_text(&document.text());
         }
@@ -227,7 +234,7 @@ impl AardvarkApplication {
         let bytes = self.imp().document.save_incremental();
         let tx = self.imp().tx.clone();
         glib::spawn_future_local(async move {
-            tx.send(bytes)
+            tx.send(FromApp::HandleMessage(bytes))
                 .await
                 .expect("sending message to networking backend");
         });

aardvark-app/src/window.rs

@@ -67,6 +67,8 @@ mod imp {
             let window = self.obj().clone();
             let dialog = self.open_document_dialog.clone();
             self.open_document_button.connect_clicked(move |_| {
+                // @TODO: Send `FromApp::SubscribeToDocument` message to network containing the
+                // share code and wait for the document to be joined.
                 dialog.present(Some(&window));
             });
         }

aardvark-node/Cargo.toml

@@ -7,13 +7,20 @@ edition = "2021"
 anyhow = "1.0.94"
 async-trait = "0.1.83"
 ciborium = "0.2.2"
+futures-io = "0.3.31"
+futures-sink = "0.3.31"
+futures-util = "0.3.31"
 iroh-gossip = "0.25.0"
-p2panda-core = "0.1.0"
-p2panda-discovery = { version = "0.1.0", features = ["mdns"] }
-p2panda-net = "0.1.0"
-p2panda-store = "0.1.0"
-p2panda-stream = "0.1.0"
-p2panda-sync = { version = "0.1.0", features = ["log-sync"] }
+p2panda-core = { git = "https://github.com/p2panda/p2panda", rev = "a174565f25c40603cfa3eb48fcde97918f23dbc8" }
+p2panda-discovery = { git = "https://github.com/p2panda/p2panda", rev = "a174565f25c40603cfa3eb48fcde97918f23dbc8", features = [
+    "mdns",
+] }
+p2panda-net = { git = "https://github.com/p2panda/p2panda", rev = "a174565f25c40603cfa3eb48fcde97918f23dbc8" }
+p2panda-store = { git = "https://github.com/p2panda/p2panda", rev = "a174565f25c40603cfa3eb48fcde97918f23dbc8" }
+p2panda-stream = { git = "https://github.com/p2panda/p2panda", rev = "a174565f25c40603cfa3eb48fcde97918f23dbc8" }
+p2panda-sync = { git = "https://github.com/p2panda/p2panda", rev = "a174565f25c40603cfa3eb48fcde97918f23dbc8", features = [
+    "log-sync",
+] }
 serde = { version = "1.0.215", features = ["derive"] }
-tokio = { version = "1.42.0", features = ["full"] }
+tokio = { version = "1.42.0", features = ["rt"] }
 tokio-stream = "0.1.17"

aardvark-node/src/document.rs

@@ -0,0 +1,64 @@
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock, RwLockWriteGuard};
+
+use async_trait::async_trait;
+use p2panda_core::PublicKey;
+use p2panda_sync::log_sync::TopicLogMap;
+
+use crate::operation::LogId;
+use crate::topics::{AardvarkTopics, TextDocument};
+
+pub type ShortCode = [char; 6];
+
+#[derive(Clone, Debug)]
+pub struct TextDocumentStore {
+    inner: Arc<RwLock<TextDocumentStoreInner>>,
+}
+
+impl Default for TextDocumentStore {
+    fn default() -> Self {
+        Self {
+            inner: Arc::new(RwLock::new(TextDocumentStoreInner {
+                authors: HashMap::new(),
+            })),
+        }
+    }
+}
+
+impl TextDocumentStore {
+    pub fn write(&self) -> RwLockWriteGuard<TextDocumentStoreInner> {
+        self.inner.write().expect("acquire write lock")
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct TextDocumentStoreInner {
+    pub authors: HashMap<PublicKey, Vec<TextDocument>>,
+}
+
+#[async_trait]
+impl TopicLogMap<AardvarkTopics, LogId> for TextDocumentStore {
+    async fn get(&self, topic: &AardvarkTopics) -> Option<HashMap<PublicKey, Vec<LogId>>> {
+        let text_document = match topic {
+            // When discovering documents we don't want any sync sessions to occur, this is a
+            // little hack to make sure that is the case, as if both peers resolve a topic to
+            // "None" then the sync session will naturally end.
+            AardvarkTopics::DiscoveryCode(_) => return None,
+            AardvarkTopics::TextDocument(text_document) => text_document,
+        };
+
+        let authors = &self.inner.read().unwrap().authors;
+        let mut result = HashMap::<PublicKey, Vec<LogId>>::new();
+
+        for (public_key, documents) in authors {
+            if documents.contains(text_document) {
+                result
+                    .entry(*public_key)
+                    .and_modify(|logs| logs.push(text_document.hash()))
+                    .or_insert(vec![text_document.hash()]);
+            }
+        }
+
+        Some(result)
+    }
+}

aardvark-node/src/lib.rs

@@ -1,2 +1,4 @@
+pub mod document;
 pub mod network;
-pub mod operation;
+pub mod operation;
+pub mod topics;