feat(walletconnect): walletconnect integration

mm2src/coins/eth/wallet_connect.rs

@@ -133,6 +133,8 @@ impl WalletConnectOps for EthCoin {
             wc.send_session_request_and_wait(&chain_id, WcRequestMethods::EthSendTransaction, tx_json, Ok)
                 .await?
         };
+
+        println!("TX_JSON: {tx_hash:?}");
         let tx_hash = tx_hash.strip_prefix("0x").unwrap_or(&tx_hash);
         let maybe_signed_tx = {
             self.wait_for_tx_appears_on_rpc(H256::from_slice(&hex::decode(tx_hash)?), WAIT_RPC_TIMEOUT_SECS, 1.)

mm2src/kdf_walletconnect/Cargo.toml

@@ -19,7 +19,6 @@ futures = { version = "0.3", package = "futures", features = [
     "async-await",
 ] }
 hkdf = "0.12.4"
-dashmap = "6.1.0"
 mm2_core = { path = "../mm2_core" }
 mm2_db = { path = "../mm2_db" }
 mm2_err_handle = { path = "../mm2_err_handle" }

mm2src/kdf_walletconnect/src/lib.rs

@@ -44,6 +44,7 @@ use std::ops::Deref;
 use std::{sync::Arc, time::Duration};
 use storage::SessionStorageDb;
 use storage::WalletConnectStorageOps;
+use tokio::time::timeout;
 use wc_common::{decode_and_decrypt_type0, encrypt_and_encode, EnvelopeType, SymKey};
 
 const PUBLISH_TIMEOUT_SECS: f64 = 6.;
@@ -286,8 +287,14 @@ impl WalletConnectCtxImpl {
             pairing_topics.push(pairing_topic);
         }
 
-        let all_topics: Vec<_> = valid_topics.into_iter().chain(pairing_topics.into_iter()).collect();
-        self.client.batch_subscribe(all_topics).await?;
+        let all_topics = valid_topics
+            .into_iter()
+            .chain(pairing_topics.into_iter())
+            .collect::<Vec<_>>();
+
+        if !all_topics.is_empty() {
+            self.client.batch_subscribe(all_topics).await?;
+        }
 
         Ok(())
     }
@@ -535,7 +542,11 @@ impl WalletConnectCtxImpl {
         let ttl = request.irn_metadata().ttl;
         self.publish_request(&active_topic, request).await?;
 
-        if let Ok(Some(resp)) = self.message_rx.lock().await.next().timeout_secs(ttl as f64).await {
+        if let Ok(Some(resp)) = timeout(Duration::from_secs(ttl), async {
+            self.message_rx.lock().await.next().await
+        })
+        .await
+        {
             let result = resp.mm_err(WalletConnectError::InternalError)?;
             if let ResponseParamsSuccess::Arbitrary(data) = result.data {
                 let data = serde_json::from_value::<T>(data)?;

mm2src/kdf_walletconnect/src/session/mod.rs

@@ -7,9 +7,6 @@ use crate::{error::WalletConnectError, WalletConnectCtxImpl};
 
 use chrono::Utc;
 use common::log::info;
-use dashmap::mapref::multiple::RefMulti;
-use dashmap::mapref::one::RefMut;
-use dashmap::DashMap;
 use derive_more::Display;
 use key::SessionKey;
 use mm2_err_handle::prelude::{MmError, MmResult};
@@ -21,9 +18,9 @@ use relay_rpc::{domain::SubscriptionId,
                 rpc::params::{session::ProposeNamespaces, session_settle::Controller, Metadata, Relay}};
 use serde::{Deserialize, Deserializer, Serialize};
 use serde_json::Value;
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::fmt::Debug;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
 use wc_common::SymKey;
 
 pub(crate) const FIVE_MINUTES: u64 = 5 * 60;
@@ -182,7 +179,7 @@ struct SessionManagerImpl {
     /// The currently active session topic.
     active_topic: Mutex<Option<Topic>>,
     /// A thread-safe map of sessions indexed by topic.
-    sessions: DashMap<Topic, Session>,
+    sessions: Arc<RwLock<HashMap<Topic, Session>>>,
     pub(crate) storage: SessionStorageDb,
 }
 
@@ -216,6 +213,14 @@ impl SessionManager {
         )
     }
 
+    pub(crate) fn read(&self) -> RwLockReadGuard<HashMap<Topic, Session>> {
+        self.0.sessions.read().expect("read shouldn't fail")
+    }
+
+    pub(crate) fn write(&self) -> RwLockWriteGuard<HashMap<Topic, Session>> {
+        self.0.sessions.write().expect("read shouldn't fail")
+    }
+
     pub(crate) fn storage(&self) -> &SessionStorageDb { &self.0.storage }
 
     /// Get active session topic or return error if no session has been activated.
@@ -236,7 +241,7 @@ impl SessionManager {
         // set active session topic.
         *self.0.active_topic.lock().unwrap() = Some(session.topic.clone());
         // insert session
-        self.0.sessions.insert(session.topic.clone(), session);
+        self.write().insert(session.topic.clone(), session);
     }
 
     /// Removes session corresponding to the specified topic from the session store.
@@ -246,12 +251,12 @@ impl SessionManager {
         let mut active_topic = self.0.active_topic.lock().unwrap();
 
         // Remove the session and get the removed session (if any)
-        let removed_session = self.0.sessions.remove(topic).map(|(_, session)| session);
+        let removed_session = self.write().remove(topic);
 
         // Update active topic if necessary
         if active_topic.as_ref() == Some(topic) {
             // If the deleted session was the active one, find a new active session
-            *active_topic = self.0.sessions.iter().next().map(|session| session.topic.clone());
+            *active_topic = self.read().iter().next().map(|(topic, session)| topic.clone());
 
             if let Some(new_active_topic) = active_topic.as_ref() {
                 info!("[{new_active_topic}] New session with topic activated!");
@@ -275,12 +280,7 @@ impl SessionManager {
     }
 
     /// Retrieves a cloned session associated with a given topic.
-    pub fn get_session(&self, topic: &Topic) -> Option<Session> { self.0.sessions.get(topic).map(|s| s.clone()) }
-
-    /// Retrieves a mutable reference to the session associated with a given topic.
-    pub(crate) fn get_session_mut(&self, topic: &Topic) -> Option<RefMut<'_, Topic, Session>> {
-        self.0.sessions.get_mut(topic)
-    }
+    pub fn get_session(&self, topic: &Topic) -> Option<Session> { self.read().get(topic).cloned() }
 
     /// returns an `option<session>` containing the active session if it exists; otherwise, returns `none`.
     pub fn get_session_active(&self) -> Option<Session> {
@@ -294,16 +294,16 @@ impl SessionManager {
 
     /// Retrieves all sessions(active and inactive)
     pub fn get_sessions(&self) -> impl Iterator<Item = SessionRpcInfo> {
-        self.0.sessions.clone().into_iter().map(|(_, session)| session.into())
+        self.read().clone().into_values().map(|session| session.into())
     }
 
-    pub(crate) fn get_sessions_full(&self) -> impl Iterator<Item = RefMulti<Topic, Session>> { self.0.sessions.iter() }
+    pub(crate) fn get_sessions_full(&self) -> impl Iterator<Item = Session> { self.read().clone().into_values() }
 
     /// Updates the expiry time of the session associated with the given topic to the specified timestamp.
     /// If the session does not exist, this method does nothing.
     pub(crate) fn extend_session(&self, topic: &Topic, till: u64) {
         info!("[{topic}] Extending session with topic");
-        if let Some(mut session) = self.0.sessions.get_mut(topic) {
+        if let Some(mut session) = self.write().get_mut(topic) {
             session.extend(till);
         }
     }

mm2src/kdf_walletconnect/src/session/rpc/event.rs

@@ -51,7 +51,8 @@ pub async fn handle_session_event(
             } else {
                 {
                     ctx.session_manager
-                        .get_session_mut(topic)
+                        .write()
+                        .get_mut(topic)
                         .ok_or(MmError::new(WalletConnectError::SessionError(
                             "No active WalletConnect session found".to_string(),
                         )))?

mm2src/kdf_walletconnect/src/session/rpc/settle.rs

@@ -3,7 +3,7 @@ use crate::storage::WalletConnectStorageOps;
 use crate::{error::WalletConnectError, WalletConnectCtxImpl};
 
 use common::log::{debug, info};
-use mm2_err_handle::prelude::{MapMmError, MmResult};
+use mm2_err_handle::prelude::{MapMmError, MmError, MmResult};
 use relay_rpc::domain::Topic;
 use relay_rpc::rpc::params::session_settle::SessionSettleRequest;
 
@@ -40,27 +40,35 @@ pub(crate) async fn reply_session_settle_request(
     settle: SessionSettleRequest,
 ) -> MmResult<(), WalletConnectError> {
     {
-        let session = ctx.session_manager.get_session_mut(topic);
-        if let Some(mut session) = session {
-            session.namespaces = settle.namespaces.0;
-            session.controller = settle.controller.clone();
-            session.relay = settle.relay;
-            session.expiry = settle.expiry;
-
-            if let Some(value) = settle.session_properties {
-                let session_properties = serde_json::from_str::<SessionProperties>(&value.to_string())?;
-                session.session_properties = Some(session_properties);
-            }
+        let mut session = ctx.session_manager.write();
+        let Some(session) = session.get_mut(topic) else {
+            return MmError::err(WalletConnectError::SessionError(format!("No session found for topic: {topic}")));
+        };
+        session.namespaces = settle.namespaces.0;
+        session.controller = settle.controller.clone();
+        session.relay = settle.relay;
+        session.expiry = settle.expiry;
 
-            // Update storage session.
-            ctx.session_manager
-                .storage()
-                .update_session(&session)
-                .await
-                .mm_err(|err| WalletConnectError::StorageError(err.to_string()))?;
+        if let Some(value) = settle.session_properties {
+            let session_properties = serde_json::from_str::<SessionProperties>(&value.to_string())?;
+            session.session_properties = Some(session_properties);
         };
-    }
-    info!("Session successfully settled for topic: {:?}", topic);
+    };
+
+    //  Update storage session.
+    let session = ctx
+        .session_manager
+        .get_session(topic)
+        .ok_or(MmError::new(WalletConnectError::SessionError(format!(
+            "session not foun topic: {topic}"
+        ))))?;
+    ctx.session_manager
+        .storage()
+        .update_session(&session)
+        .await
+        .mm_err(|err| WalletConnectError::StorageError(err.to_string()))?;
+
+    info!("[{topic}] Session successfully settled for topic");
 
     // Delete other sessions with same controller
     // TODO: we might not want to do this!

mm2src/kdf_walletconnect/src/session/rpc/update.rs

@@ -15,24 +15,33 @@ pub(crate) async fn reply_session_update_request(
     update: SessionUpdateRequest,
 ) -> MmResult<(), WalletConnectError> {
     {
-        if let Some(mut session) = ctx.session_manager.get_session_mut(topic) {
-            update
-                .namespaces
-                .caip2_validate()
-                .map_to_mm(|err| WalletConnectError::InternalError(err.to_string()))?;
-            //TODO: session.namespaces.supported(update.namespaces.0)
-            session.namespaces = update.namespaces.0;
-            //  Update storage session.
-            ctx.session_manager
-                .storage()
-                .update_session(&session)
-                .await
-                .mm_err(|err| WalletConnectError::StorageError(err.to_string()))?;
-
-            info!("Updated extended, info: {:?}", session.topic);
+        let mut session = ctx.session_manager.write();
+        let Some(session) = session.get_mut(topic) else {
+            return MmError::err(WalletConnectError::SessionError(format!("No session found for topic: {topic}")));
         };
+        update
+            .namespaces
+            .caip2_validate()
+            .map_to_mm(|err| WalletConnectError::InternalError(err.to_string()))?;
+        //TODO: session.namespaces.supported(update.namespaces.0)
+        session.namespaces = update.namespaces.0;
+        let session = session;
+        info!("Updated extended, info: {:?}", session.topic);
     }
 
+    //  Update storage session.
+    let session = ctx
+        .session_manager
+        .get_session(topic)
+        .ok_or(MmError::new(WalletConnectError::SessionError(format!(
+            "session not foun topic: {topic}"
+        ))))?;
+    ctx.session_manager
+        .storage()
+        .update_session(&session)
+        .await
+        .mm_err(|err| WalletConnectError::StorageError(err.to_string()))?;
+
     let param = ResponseParamsSuccess::SessionUpdate(true);
     ctx.publish_response_ok(topic, param, message_id).await?;