refactor: runtime agnostic

Cargo.toml

@@ -43,13 +43,15 @@ default-members = [
 aes = "0.8.3"
 aes-gcm = { version = "0.10.2", default-features = false, features = ["aes", "alloc", "std"] }
 anyhow = { version = "1.0", default-features = false }
-async-channel = { version = "2.5.0", default-features = false }
+async-channel = { version = "2.5.0", default-features = false, features = ["std"] }
+async-lock = { version = "3", default-features = false }
 async-trait = "0.1.89"
 base64 = { version = "0.22.1", default-features = false, features = ["alloc"] }
 bytemuck = { version = "1.22", default-features = false }
 bytes = { version = "1.5", default-features = false, features = ["serde"] }
 chrono = { version = "0.4", default-features = false }
 ctr = { version = "0.9", default-features = false }
+event-listener = { version = "5", default-features = false }
 flate2 = { version = "1.1.5", default-features = false, features = ["zlib-rs"] }
 futures = { version = "0.3", default-features = false, features = ["alloc", "async-await"] }
 hex = { version = "0.4", default-features = false, features = ["alloc"] }
@@ -78,10 +80,17 @@ wacore-noise = { path = "./wacore/noise", version = "0.3.0" }
 waproto = { path = "./waproto", version = "0.3.0" }
 
 [features]
-debug-diagnostics = []
-debug-snapshots = []
+debug-diagnostics = ["wacore/debug-diagnostics"]
+debug-snapshots = ["wacore/debug-snapshots"]
 danger-skip-tls-verify = ["whatsapp-rust-tokio-transport?/danger-skip-tls-verify"]
-default = ["simd", "sqlite-storage", "tokio-transport", "ureq-client", "tokio-native", "signal"]
+default = [
+    "simd",
+    "sqlite-storage",
+    "tokio-transport",
+    "ureq-client",
+    "tokio-native",
+    "signal",
+]
 simd = ["wacore/simd"]
 ureq-client = ["dep:whatsapp-rust-ureq-http-client"]
 tokio-transport = ["dep:whatsapp-rust-tokio-transport"]
@@ -92,12 +101,15 @@ tokio-native = ["tokio/rt-multi-thread"]
 [dependencies]
 anyhow = { workspace = true }
 async-channel = { workspace = true }
+async-lock = { workspace = true }
 async-trait = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true, features = ["clock", "serde"] }
 dashmap = "6.1.0"
 env_logger = { version = "0.11", default-features = false }
+event-listener = { workspace = true }
+futures = { workspace = true, features = ["std"] }
 hex = { workspace = true }
 log = { workspace = true }
 moka = { version = "0.12.12", features = ["future"] }

examples/benchmark.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use wacore::proto_helpers::MessageExt;
 use wacore::types::events::Event;
 use waproto::whatsapp as wa;
+use whatsapp_rust::TokioRuntime;
 use whatsapp_rust::bot::{Bot, MessageContext};
 use whatsapp_rust::store::SqliteStore;
 use whatsapp_rust_tokio_transport::TokioWebSocketTransportFactory;
@@ -48,7 +49,8 @@ fn main() {
         let builder = Bot::builder()
             .with_backend(backend)
             .with_transport_factory(transport_factory)
-            .with_http_client(http_client);
+            .with_http_client(http_client)
+            .with_runtime(TokioRuntime);
 
         let mut bot = builder
             .on_event(move |event, client| async move {

src/appstate_sync.rs

@@ -2,10 +2,10 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use anyhow::{Result, anyhow};
+use async_lock::Mutex;
 use async_trait::async_trait;
 use prost::Message;
 use thiserror::Error;
-use tokio::sync::Mutex;
 use wacore::appstate::hash::HashState;
 use wacore::appstate::keys::ExpandedAppStateKeys;
 use wacore::appstate::patch_decode::{PatchList, WAPatchName, parse_patch_list, parse_patch_lists};
@@ -32,12 +32,14 @@ pub enum AppStateSyncError {
 #[derive(Clone)]
 pub struct AppStateProcessor {
     pub(crate) backend: Arc<dyn Backend>,
+    pub(crate) runtime: Arc<dyn wacore::runtime::Runtime>,
     key_cache: Arc<Mutex<HashMap<String, Arc<ExpandedAppStateKeys>>>>,
 }
 
 impl AppStateProcessor {
-    pub fn new(backend: Arc<dyn Backend>) -> Self {
+    pub fn new(backend: Arc<dyn Backend>, runtime: Arc<dyn wacore::runtime::Runtime>) -> Self {
         Self {
+            runtime,
             backend,
             key_cache: Arc::new(Mutex::new(HashMap::new())),
         }
@@ -223,7 +225,7 @@ impl AppStateProcessor {
             let collection_name_owned = collection_name.to_string();
 
             // Offload CPU-intensive snapshot processing to a blocking thread
-            let result = tokio::task::spawn_blocking(move || {
+            let result = wacore::runtime::blocking(&*self.runtime, move || {
                 let get_keys = |key_id: &[u8]| -> Result<
                     ExpandedAppStateKeys,
                     wacore::appstate::AppStateError,
@@ -248,7 +250,6 @@ impl AppStateProcessor {
                 Ok::<_, wacore::appstate::AppStateError>((result, snapshot_state))
             })
             .await
-            .map_err(|e| anyhow!("Blocking task failed: {}", e))?
             .map_err(|e| anyhow!("{}", e))?;
 
             let (snapshot_result, snapshot_state) = result;
@@ -309,7 +310,7 @@ impl AppStateProcessor {
             let coll = collection_name_owned.clone();
 
             // Offload CPU-intensive patch processing to a blocking thread
-            let result = tokio::task::spawn_blocking(move || {
+            let result = wacore::runtime::blocking(&*self.runtime, move || {
                 let get_keys = |key_id: &[u8]| -> Result<
                     ExpandedAppStateKeys,
                     wacore::appstate::AppStateError,
@@ -338,7 +339,6 @@ impl AppStateProcessor {
                 )
             })
             .await
-            .map_err(|e| anyhow!("Blocking task failed: {}", e))?
             .map_err(|e| anyhow!("{}", e))?;
 
             // Update local state with the result from the blocking task
@@ -769,7 +769,8 @@ mod tests {
     #[tokio::test]
     async fn test_process_patch_list_handles_set_overwrite_correctly() {
         let backend = Arc::new(MockBackend::default());
-        let processor = AppStateProcessor::new(backend.clone());
+        let processor =
+            AppStateProcessor::new(backend.clone(), Arc::new(crate::runtime_impl::TokioRuntime));
         let collection_name = WAPatchName::Regular;
         let index_mac = vec![1; 32];
         let key_id_bytes = b"test_key_id".to_vec();