feat: add Signal protocol feature API and consolidate internal helpers

src/client.rs

@@ -450,6 +450,10 @@ pub struct Client {
     /// Weak self-reference for spawning background tasks from `&self` methods.
     /// Initialized after `Arc::new(this)` in the constructor.
     pub(crate) self_weak: std::sync::OnceLock<std::sync::Weak<Client>>,
+
+    /// When true, emit `Event::RawNode` for every decoded stanza before router dispatch.
+    /// Default false — only enable when external consumers need raw protocol access.
+    raw_node_forwarding: AtomicBool,
 }
 
 impl Client {
@@ -704,6 +708,7 @@ impl Client {
             skip_history_sync: AtomicBool::new(false),
             cache_config,
             self_weak: std::sync::OnceLock::new(),
+            raw_node_forwarding: AtomicBool::new(false),
         };
 
         let arc = Arc::new(this);
@@ -799,6 +804,68 @@ impl Client {
         self.core.event_bus.add_handler(handler);
     }
 
+    /// Enable or disable raw node forwarding.
+    /// When enabled, `Event::RawNode` is emitted for every decoded stanza before
+    /// the stanza router dispatches it. Only enable when external consumers need
+    /// raw protocol access (e.g. voice call stanzas).
+    pub fn set_raw_node_forwarding(&self, enabled: bool) {
+        self.raw_node_forwarding.store(enabled, Ordering::Relaxed);
+    }
+
+    /// Build a [`SignalProtocolStoreAdapter`] from the current device state and signal cache.
+    pub(crate) async fn signal_adapter(
+        &self,
+    ) -> crate::store::signal_adapter::SignalProtocolStoreAdapter {
+        let device_store = self.persistence_manager.get_device_arc().await;
+        self.signal_adapter_from(device_store)
+    }
+
+    /// Build a [`SignalProtocolStoreAdapter`] from a pre-fetched device arc.
+    pub(crate) fn signal_adapter_from(
+        &self,
+        device_store: Arc<async_lock::RwLock<crate::store::Device>>,
+    ) -> crate::store::signal_adapter::SignalProtocolStoreAdapter {
+        crate::store::signal_adapter::SignalProtocolStoreAdapter::new(
+            device_store,
+            self.signal_cache.clone(),
+        )
+    }
+
+    /// Get the per-address session mutex from the lock cache.
+    pub(crate) async fn session_lock_for(
+        &self,
+        signal_addr_str: &str,
+    ) -> Arc<async_lock::Mutex<()>> {
+        self.session_locks
+            .get_with_by_ref(signal_addr_str, async {
+                Arc::new(async_lock::Mutex::new(()))
+            })
+            .await
+    }
+
+    /// Get the active noise socket, or error if not connected.
+    pub(crate) async fn get_noise_socket(
+        &self,
+    ) -> Result<Arc<crate::socket::noise_socket::NoiseSocket>, ClientError> {
+        self.noise_socket
+            .lock()
+            .await
+            .clone()
+            .ok_or(ClientError::NotConnected)
+    }
+
+    /// Send pre-marshaled plaintext bytes through the noise socket.
+    pub async fn send_raw_bytes(&self, plaintext: Vec<u8>) -> Result<(), ClientError> {
+        let noise_socket = self.get_noise_socket().await?;
+        let encrypted_buf = Vec::with_capacity(plaintext.len() + 32);
+        noise_socket
+            .encrypt_and_send(plaintext, encrypted_buf)
+            .await?;
+        self.last_data_sent_ms
+            .store(wacore::time::now_millis().max(0) as u64, Ordering::Relaxed);
+        Ok(())
+    }
+
     /// Register a chatstate handler which will be invoked when a `<chatstate>` stanza is received.
     ///
     /// The handler receives a `ChatStateEvent` with the parsed chat state information.
@@ -1349,10 +1416,9 @@ impl Client {
         self: &Arc<Self>,
         encrypted_frame: &bytes::Bytes,
     ) -> Option<wacore_binary::node::Node> {
-        let noise_socket_arc = { self.noise_socket.lock().await.clone() };
-        let noise_socket = match noise_socket_arc {
-            Some(s) => s,
-            None => {
+        let noise_socket = match self.get_noise_socket().await {
+            Ok(s) => s,
+            Err(_) => {
                 log::error!("Cannot process frame: not connected (no noise socket)");
                 return None;
             }
@@ -1519,6 +1585,14 @@ impl Client {
             }
         }
 
+        // Emit raw node for external observers (e.g. voice call handlers)
+        // Zero-cost when disabled: just an atomic load
+        if self.raw_node_forwarding.load(Ordering::Relaxed) {
+            self.core
+                .event_bus
+                .dispatch(&Event::RawNode(Arc::clone(&node)));
+        }
+
         // Dispatch to appropriate handler using the router
         // Clone Arc (cheap - just reference count) not the Node itself
         if !self
@@ -3329,39 +3403,18 @@ impl Client {
     }
 
     pub async fn send_node(&self, node: Node) -> Result<(), ClientError> {
-        let noise_socket_arc = { self.noise_socket.lock().await.clone() };
-        let noise_socket = match noise_socket_arc {
-            Some(socket) => socket,
-            None => return Err(ClientError::NotConnected),
-        };
-
         debug!(target: "Client/Send", "{}", DisplayableNode(&node));
         if self.sent_node_waiter_count.load(Ordering::Acquire) > 0 {
             self.resolve_sent_node_waiters(&Arc::new(node.clone()));
         }
 
         let mut plaintext_buf = Vec::with_capacity(1024);
-
         if let Err(e) = wacore_binary::marshal::marshal_to(&node, &mut plaintext_buf) {
             error!("Failed to marshal node: {e:?}");
             return Err(SocketError::Crypto("Marshal error".to_string()).into());
         }
 
-        // Size based on plaintext + encryption overhead (16 byte tag + 3 byte frame header)
-        let encrypted_buf = Vec::with_capacity(plaintext_buf.len() + 32);
-
-        if let Err(e) = noise_socket
-            .encrypt_and_send(plaintext_buf, encrypted_buf)
-            .await
-        {
-            return Err(e.into());
-        }
-
-        // WA Web: callStanza → deadSocketTimer.onOrBefore(deadSocketTime, socketId)
-        self.last_data_sent_ms
-            .store(wacore::time::now_millis().max(0) as u64, Ordering::Relaxed);
-
-        Ok(())
+        self.send_raw_bytes(plaintext_buf).await
     }
 
     pub(crate) async fn update_push_name_and_notify(self: &Arc<Self>, new_name: String) {

src/client/sessions.rs

@@ -198,11 +198,7 @@ impl Client {
             .fetch_pre_keys(jids, Some(wacore::iq::prekeys::PreKeyFetchReason::Identity))
             .await?;
 
-        let device_store = self.persistence_manager.get_device_arc().await;
-        let mut adapter = crate::store::signal_adapter::SignalProtocolStoreAdapter::new(
-            device_store,
-            self.signal_cache.clone(),
-        );
+        let mut adapter = self.signal_adapter().await;
 
         let mut success_count = 0;
         let mut missing_count = 0;
@@ -213,12 +209,7 @@ impl Client {
                 let signal_addr = jid.to_protocol_address();
 
                 // Acquire per-sender session lock to prevent race with concurrent message decryption.
-                let session_mutex = self
-                    .session_locks
-                    .get_with_by_ref(signal_addr.as_str(), async {
-                        std::sync::Arc::new(async_lock::Mutex::new(()))
-                    })
-                    .await;
+                let session_mutex = self.session_lock_for(signal_addr.as_str()).await;
                 let _session_guard = session_mutex.lock().await;
 
                 match process_prekey_bundle(

src/features/mod.rs

@@ -10,6 +10,7 @@ pub(crate) mod newsletter;
 mod polls;
 mod presence;
 mod profile;
+mod signal;
 pub(crate) mod status;
 mod tctoken;
 
@@ -49,4 +50,6 @@ pub use profile::{Profile, SetProfilePictureResponse};
 
 pub use status::{Status, StatusPrivacySetting, StatusSendOptions};
 
+pub use signal::Signal;
+
 pub use tctoken::TcToken;