[WIP] Implement sending messages in backend

Author: ChronosXYZ

Cargo.lock

@@ -40,3 +40,5 @@ identicon-rs = "4.0.1"
 adw = { package = "libadwaita", version = "0.3.0", features = ["gtk_v4_6", "v1_3"] }
 gtk = { package = "gtk4",version = "0.6.6", features = ["v4_10"] }
 mail-parser = "0.8.2"
+strum = { version = "0.24", features = ["derive"] }
+emailmessage = "0.2.2"

Cargo.toml

@@ -0,0 +1,13 @@
+version: "3"
+
+services:
+  devzone:
+    image: "ubuntu:lunar"
+    stdin_open: true
+    tty: true
+    volumes:
+      - ../:/app/bitmessage-rs:rw
+    network_mode: host
+    environment:
+      - DISPLAY
+    entrypoint: /bin/bash

devzone/docker-compose.yaml

@@ -51,6 +51,7 @@
             pango
             gtk4
             libadwaita
+            adwaita-icon-theme
             openssl
             sqlite
             (if system == "aarch64-darwin" then
@@ -73,6 +74,7 @@
             gdk-pixbuf
             pango
             gtk4
+            gnome.adwaita-icon-theme
             libadwaita
             openssl
             sqlite

flake.nix

@@ -79,7 +79,7 @@ impl Object {
         object
     }
 
-    pub fn send(mut self, mut worker_sink: mpsc::Sender<WorkerCommand>) {
+    pub fn do_proof_of_work(mut self, mut worker_sink: mpsc::Sender<WorkerCommand>) {
         let target = pow::get_pow_target(
             &self,
             pow::NETWORK_MIN_NONCE_TRIALS_PER_BYTE,

src/network/messages.rs

@@ -1,12 +1,17 @@
+use emailmessage::{header, Message, SinglePart};
 use std::error::Error;
 
+use chrono::Utc;
 use futures::{
     channel::{mpsc, oneshot},
     SinkExt,
 };
 use libp2p::{Multiaddr, PeerId};
 
-use crate::{network::address::Address, repositories::sqlite::models};
+use crate::{
+    network::address::Address,
+    repositories::sqlite::models::{self, MessageStatus},
+};
 
 use super::worker::{Folder, WorkerCommand};
 
@@ -117,4 +122,35 @@ impl NodeClient {
             .expect("Sender not to be dropped")
             .expect("repo not to fail")
     }
+
+    pub async fn send_message(&mut self, from: String, to: String, title: String, body: String) {
+        let (sender, receiver) = oneshot::channel();
+        let m: Message<SinglePart<&str>> = Message::builder().subject(title).mime_body(
+            SinglePart::builder()
+                .header(header::ContentType(
+                    "text/plain; charset=utf8".parse().unwrap(),
+                ))
+                .header(header::ContentTransferEncoding::QuotedPrintable)
+                .body(&body),
+        );
+        let data = m.to_string().into_bytes();
+        let msg = models::Message {
+            hash: "".to_string(),
+            sender: from.clone(),
+            recipient: to,
+            created_at: Utc::now().naive_utc(),
+            status: MessageStatus::Unknown.to_string(),
+            signature: Vec::new(),
+            data,
+        };
+
+        self.sender
+            .send(WorkerCommand::SendMessage { msg, from, sender })
+            .await
+            .expect("Receiver not to be dropped");
+        receiver
+            .await
+            .expect("Sender not to be dropped")
+            .expect("repo not to fail")
+    }
 }

src/network/node/client.rs

@@ -1,12 +1,10 @@
-use async_std::task;
-use sha2::Digest;
 use std::{error::Error, sync::Arc};
 
 use chrono::Utc;
 use futures::{
     channel::{mpsc, oneshot},
     lock::Mutex,
-    FutureExt, SinkExt,
+    SinkExt,
 };
 
 use crate::{
@@ -31,6 +29,7 @@ pub struct Handler {
     message_repo: Box<MessageRepositorySync>,
     requested_objects: Vec<String>, // TODO periodically request missing object from every connection we have
     worker_event_sender: mpsc::Sender<WorkerCommand>,
+    pubkey_notifier_sink: mpsc::Sender<String>,
 }
 
 impl Handler {
@@ -39,13 +38,15 @@ impl Handler {
         inventory_repo: Box<InventoryRepositorySync>,
         message_repo: Box<MessageRepositorySync>,
         worker_event_sender: mpsc::Sender<WorkerCommand>,
+        pubkey_notifier_sink: mpsc::Sender<String>,
     ) -> Handler {
         Handler {
             address_repo,
             inventory_repo: Arc::new(Mutex::new(inventory_repo)),
             message_repo,
             requested_objects: Vec::new(),
             worker_event_sender,
+            pubkey_notifier_sink,
         }
     }
 
@@ -201,6 +202,8 @@ impl Handler {
             .await
             .expect("repo not to fail");
 
+        self.pubkey_notifier_sink.send(tag_str).await.unwrap();
+
         Ok(())
     }
 
@@ -241,7 +244,7 @@ impl Handler {
                     },
                     expires,
                 );
-                obj.send(self.worker_event_sender.clone());
+                obj.do_proof_of_work(self.worker_event_sender.clone());
             }
         }
 

src/network/node/handler.rs

@@ -1,18 +1,17 @@
-use async_std::task;
 use chrono::Utc;
 use diesel::{
     connection::SimpleConnection,
     r2d2::{ConnectionManager, Pool},
     SqliteConnection,
 };
 use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
-use sha2::Digest;
-use std::{borrow::Cow, error::Error, fs, io, iter, time::Duration};
+use rand::distributions::{Alphanumeric, DistString};
+use std::{borrow::Cow, collections::HashMap, error::Error, fs, io, iter, time::Duration};
 
 use directories::ProjectDirs;
 use futures::{
     channel::{mpsc, oneshot},
-    select, FutureExt, SinkExt, StreamExt,
+    select, StreamExt,
 };
 use libp2p::{
     core::upgrade::Version,
@@ -34,16 +33,17 @@ use crate::{
             BitmessageBehaviourEvent, BitmessageNetBehaviour, BitmessageProtocol,
             BitmessageProtocolCodec, BitmessageResponse,
         },
-        messages::{self, MessageCommand, MessagePayload, ObjectKind, UnencryptedMsg},
+        messages::{self, MessageCommand, MessagePayload, MsgEncoding, ObjectKind, UnencryptedMsg},
     },
-    pow,
     repositories::{
         address::AddressRepositorySync,
         inventory::InventoryRepositorySync,
         message::MessageRepositorySync,
         sqlite::{
-            address::SqliteAddressRepository, inventory::SqliteInventoryRepository,
-            message::SqliteMessageRepository, models,
+            address::SqliteAddressRepository,
+            inventory::SqliteInventoryRepository,
+            message::SqliteMessageRepository,
+            models::{self, MessageStatus},
         },
     },
 };
@@ -135,6 +135,7 @@ pub enum WorkerCommand {
     },
     SendMessage {
         msg: models::Message,
+        from: String,
         sender: oneshot::Sender<Result<(), DynError>>,
     },
 }
@@ -143,7 +144,12 @@ pub struct NodeWorker {
     local_peer_id: PeerId,
     swarm: Swarm<BitmessageNetBehaviour>,
     handler: Handler,
+    command_sender: mpsc::Sender<WorkerCommand>,
     command_receiver: mpsc::Receiver<WorkerCommand>,
+
+    pubkey_notifier: mpsc::Receiver<String>,
+    tracked_pubkeys: HashMap<String, bool>, // TODO populate it on startup
+
     pending_commands: Vec<WorkerCommand>,
     _sqlite_connection_pool: Pool<ConnectionManager<SqliteConnection>>,
     common_topic: Sha256Topic,
@@ -251,6 +257,7 @@ impl NodeWorker {
             .expect("subscription not to fail");
 
         let (sender, receiver) = mpsc::channel(0);
+        let (pubkey_notifier_sink, pubkey_notifier) = mpsc::channel(0);
         let inventory_repo = Box::new(SqliteInventoryRepository::new(pool.clone()));
         let address_repo = Box::new(SqliteAddressRepository::new(pool.clone()));
         let message_repo = Box::new(SqliteMessageRepository::new(pool.clone()));
@@ -263,7 +270,11 @@ impl NodeWorker {
                     inventory_repo.clone(),
                     message_repo.clone(),
                     sender.clone(),
+                    pubkey_notifier_sink,
                 ),
+                command_sender: sender.clone(),
+                pubkey_notifier,
+                tracked_pubkeys: HashMap::new(),
                 command_receiver: receiver,
                 pending_commands: Vec::new(),
                 _sqlite_connection_pool: pool,
@@ -384,6 +395,18 @@ impl NodeWorker {
                     .expect("receiver not to be dropped"),
             },
             WorkerCommand::NonceCalculated { obj } => {
+                match &obj.kind {
+                    ObjectKind::Msg { encrypted: _ } => self
+                        .messages_repo
+                        .update_message_status(
+                            bs58::encode(&obj.hash).into_string(),
+                            MessageStatus::Sent,
+                        )
+                        .await
+                        .unwrap(),
+                    _ => {}
+                }
+
                 self.inventory_repo
                     .store_object(bs58::encode(&obj.hash).into_string(), obj)
                     .await
@@ -468,7 +491,71 @@ impl NodeWorker {
                         .expect("receiver not to be dropped"),
                 },
             },
-            WorkerCommand::SendMessage { msg, sender } => {}
+            WorkerCommand::SendMessage {
+                mut msg,
+                from,
+                sender,
+            } => {
+                let identity = self
+                    .address_repo
+                    .get_by_ripe_or_tag(from)
+                    .await
+                    .unwrap()
+                    .unwrap();
+                let recipient: Option<Address> = self
+                    .address_repo
+                    .get_by_ripe_or_tag(msg.recipient.clone())
+                    .await
+                    .unwrap();
+                match recipient {
+                    Some(v) => {
+                        msg.status = MessageStatus::WaitingForPOW.to_string();
+                        let unenc_msg = UnencryptedMsg {
+                            behavior_bitfield: 0,
+                            sender_ripe: msg.sender.clone(),
+                            destination_ripe: msg.recipient.clone(),
+                            encoding: MsgEncoding::Simple,
+                            message: msg.data.clone(),
+                            public_encryption_key: v
+                                .public_encryption_key
+                                .unwrap()
+                                .serialize()
+                                .to_vec(),
+                            public_signing_key: v.public_signing_key.unwrap().serialize().to_vec(),
+                        };
+                        let encrypted = Self::serialize_and_encrypt_payload_pub(
+                            unenc_msg,
+                            &v.public_encryption_key.unwrap(),
+                        );
+                        let object = messages::Object::with_signing(
+                            &identity,
+                            ObjectKind::Msg { encrypted },
+                            Utc::now() + chrono::Duration::days(7),
+                        );
+
+                        msg.hash = bs58::encode(&object.hash).into_string();
+                        self.messages_repo.save_model(msg.clone()).await.unwrap();
+                        object.do_proof_of_work(self.command_sender.clone());
+                    }
+                    None => {
+                        msg.status = MessageStatus::WaitingForPubkey.to_string();
+                        // we generate random hash value, cuz we don't really know real hash value of the message at the moment, and it's not that important
+                        msg.hash = Alphanumeric.sample_string(&mut rand::thread_rng(), 16);
+                        self.messages_repo.save_model(msg.clone()).await.unwrap();
+                        // send getpubkey request
+                        let obj = messages::Object::with_signing(
+                            &identity,
+                            ObjectKind::Getpubkey {
+                                tag: Address::new(bs58::decode(msg.recipient).into_vec().unwrap())
+                                    .tag,
+                            },
+                            Utc::now() + chrono::Duration::days(7),
+                        );
+                        obj.do_proof_of_work(self.command_sender.clone());
+                    }
+                }
+                sender.send(Ok(())).unwrap();
+            }
         };
     }
 
@@ -492,11 +579,18 @@ impl NodeWorker {
                         log::debug!("Shutting down network event loop...");
                         return;
                     },
-               }
+                },
+                pubkey_notification = self.pubkey_notifier.next() => self.handle_pubkey_notification(pubkey_notification.unwrap()).await,
             }
         }
     }
 
+    async fn handle_pubkey_notification(&mut self, tag: String) {
+        if let Some(_) = self.tracked_pubkeys.get(&tag) {
+            // TODO seek for messages which haven't been sent yet because of empty public key of recipient
+        }
+    }
+
     /// When we receive IdentityInfo, if the peer supports our Kademlia protocol, we add
     /// their listen addresses to the DHT, so they will be propagated to other peers.
     fn handle_identify_event(&mut self, identify_event: identify::Event) {
@@ -541,6 +635,21 @@ impl NodeWorker {
         .unwrap();
         encrypted
     }
+
+    pub fn serialize_and_encrypt_payload_pub<T>(
+        object: T,
+        public_key: &libsecp256k1::PublicKey,
+    ) -> Vec<u8>
+    where
+        T: Serialize,
+    {
+        let encrypted = ecies::encrypt(
+            &public_key.serialize(),
+            serde_cbor::to_vec(&object).unwrap().as_ref(),
+        )
+        .unwrap();
+        encrypted
+    }
 }
 
 fn extract_peer_id_from_multiaddr(

src/network/node/worker.rs

@@ -4,7 +4,7 @@ use async_trait::async_trait;
 
 use crate::network::messages::UnencryptedMsg;
 
-use super::sqlite::models;
+use super::sqlite::models::{self, MessageStatus};
 
 #[async_trait]
 pub trait MessageRepository {
@@ -30,6 +30,12 @@ pub trait MessageRepository {
         &self,
         address: String,
     ) -> Result<Vec<models::Message>, Box<dyn Error>>;
+
+    async fn update_message_status(
+        &mut self,
+        hash: String,
+        status: MessageStatus,
+    ) -> Result<(), Box<dyn Error>>;
 }
 
 pub type MessageRepositorySync = dyn MessageRepository + Sync + Send;