feat(plugins): add extensible client architecture

examples/voip-cli/src/main.rs

@@ -1660,7 +1660,7 @@ async fn run_bot(mode: Mode) -> Result<()> {
     // accept flow, so the raw-node-forwarding crutch the old hand-rolled inbound path needed is gone.
     let manages_media = accept || target.is_some();
     let observer = Arc::new(CallObserver::new(client.clone(), accept, video, audio));
-    client.register_handler(observer.clone());
+    let _observer_subscription = client.subscribe_handler(observer.clone());
 
     if let Some(peer) = target {
         let client2 = client.clone();

src/bot.rs

@@ -1,5 +1,5 @@
 use crate::cache_config::CacheConfig;
-use crate::client::Client;
+use crate::client::{Client, ClientBuilderError};
 use crate::pair_code::PairCodeOptions;
 use crate::store::commands::DeviceCommand;
 use crate::store::error::StoreError;
@@ -87,49 +87,8 @@ pub enum BotBuilderError {
     /// Initializing the device row in the storage backend failed.
     #[error("failed to initialize the device store: {0}")]
     Store(#[from] StoreError),
-    /// An inbound durability hook was registered with a backend that does not
-    /// implement the pending-inbound buffer it requires.
-    #[error("the configured backend does not support the inbound durability hook: {0}")]
-    UnsupportedDurabilityBackend(String),
-}
-
-/// Verify the backend round-trips a pending-inbound buffer entry before we accept
-/// an inbound durability hook. A backend relying on the no-op/`Err` trait
-/// defaults fails here instead of silently looping every inbound message unacked.
-async fn probe_durability_backend(
-    backend: &std::sync::Arc<dyn Backend>,
-) -> std::result::Result<(), BotBuilderError> {
-    // A real JID (a backend may validate the format) and an id unique per probe
-    // invocation (pid + atomic counter) so concurrent builders on the same store
-    // never race on a shared probe row and false-fail.
-    use portable_atomic::{AtomicU64, Ordering};
-    static PROBE_SEQ: AtomicU64 = AtomicU64::new(0);
-    const PROBE_JID: &str = "0@s.whatsapp.net";
-    const PROBE_PAYLOAD: &[u8] = b"probe";
-    let probe_id = format!(
-        "__wa_durability_probe_{}_{}__",
-        std::process::id(),
-        PROBE_SEQ.fetch_add(1, Ordering::Relaxed)
-    );
-    let map_err = |e: StoreError| BotBuilderError::UnsupportedDurabilityBackend(e.to_string());
-    backend
-        .store_pending_inbound(PROBE_JID, PROBE_JID, &probe_id, PROBE_PAYLOAD)
-        .await
-        .map_err(map_err)?;
-    let got = backend
-        .get_pending_inbound(PROBE_JID, PROBE_JID, &probe_id)
-        .await
-        .map_err(map_err)?;
-    backend
-        .delete_pending_inbound(PROBE_JID, PROBE_JID, &probe_id)
-        .await
-        .map_err(map_err)?;
-    if got.as_deref() != Some(PROBE_PAYLOAD) {
-        return Err(BotBuilderError::UnsupportedDurabilityBackend(
-            "pending-inbound buffer did not round-trip".to_string(),
-        ));
-    }
-    Ok(())
+    #[error(transparent)]
+    Client(#[from] ClientBuilderError),
 }
 
 /// `message` is `Arc` so cloning the context across spawned tasks only bumps a
@@ -631,14 +590,15 @@ impl Bot {
             client
                 .core
                 .event_bus
-                .add_handler(Arc::new(CallbackBusAdapter::new(
+                .subscribe_handler(Arc::new(CallbackBusAdapter::new(
                     client.clone(),
                     event_handlers,
                     event_delivery,
-                )));
+                )))
+                .detach();
         }
         for handler in raw_handlers {
-            client.core.event_bus.add_handler(handler);
+            client.core.event_bus.subscribe_handler(handler).detach();
         }
 
         // If pair code options are set, spawn a task to request pair code after socket is ready
@@ -1278,18 +1238,8 @@ impl BotBuilder<Provided, Provided, Provided, Provided> {
             unreachable!("typestate guarantees all required fields are Provided")
         };
 
-        // Instrument the runtime before anything spawns through it, so every
-        // internal task (noise sender, saver, workers) reports to the hook.
-        // Default (None): the original runtime is used untouched. The Bot
-        // keeps its own copy for the `run()` path (see the field doc).
         let task_instrument = self.task_instrument;
         let alloc_meter = self.alloc_meter;
-        let runtime: Arc<dyn Runtime> = match task_instrument.clone() {
-            Some(instrument) => {
-                Arc::new(wacore::stats::InstrumentedRuntime::new(runtime, instrument))
-            }
-            None => runtime,
-        };
 
         // Note: For multi-account mode, create the backend with SqliteStore::new_for_device()
         // before passing it to with_backend_arc()
@@ -1331,57 +1281,37 @@ impl BotBuilder<Provided, Provided, Provided, Provided> {
         }
 
         info!("Creating client...");
-        let (client, sync_task_receiver) = Client::new_with_cache_config(
-            runtime.clone(),
-            persistence_manager.clone(),
-            transport_factory,
-            http_client,
-            self.override_version,
-            self.cache_config,
-        )
-        .await;
-
-        let saver_handle = persistence_manager.run_background_saver(
-            runtime,
-            std::time::Duration::from_secs(30),
-            client.shutdown_signal(),
-        );
-        // Tie the saver task to Arc<Client> so extracting client() and outliving
-        // Bot keeps periodic persistence alive. Client::drop on the last Arc
-        // drops the AbortHandle and aborts the task.
-        let _ = client.saver_handle.set(saver_handle);
-
-        // Typed alloc-meter handle for resource_report (its poll hooks are
-        // already wired via task_instrument above).
-        if let Some(meter) = alloc_meter {
-            let _ = client.alloc_meter.set(meter);
+        let mut client_builder = Client::builder()
+            .with_runtime_arc(runtime)
+            .with_persistence_manager(persistence_manager)
+            .with_transport_factory_arc(transport_factory)
+            .with_http_client_arc(http_client)
+            .with_cache_config(self.cache_config)
+            .with_custom_enc_handlers(self.custom_enc_handlers)
+            .with_skip_history_sync(self.skip_history_sync)
+            .with_background_saver_interval(std::time::Duration::from_secs(30));
+
+        if let Some(version) = self.override_version {
+            client_builder = client_builder.with_version_override(version);
         }
-
-        // Register custom enc handlers. Immutable after build, so set the whole
-        // map once; the receive hot path then reads it lock-free.
-        let _ = client.custom_enc_handlers.set(self.custom_enc_handlers);
-
-        // Inbound durability hook (opt-in). Immutable after build; the receive
-        // path reads it lock-free. Probe the backend first: a backend that does
-        // not implement the pending-inbound buffer would otherwise leave every
-        // inbound message unacked and looping forever at runtime, so reject it
-        // here with a clear error instead.
         if let Some(hook) = self.inbound_durability_hook {
-            probe_durability_backend(&client.persistence_manager.backend()).await?;
-            let _ = client.inbound_durability_hook.set(hook);
+            client_builder = client_builder.with_inbound_durability_hook_arc(hook);
         }
-
-        if self.skip_history_sync {
-            client.set_skip_history_sync(true);
-        }
-
         if let Some(count) = self.wanted_pre_key_count {
-            client.set_wanted_pre_key_count(count);
+            client_builder = client_builder.with_wanted_pre_key_count(count);
         }
-
         if let Some((burst, refill_per_min)) = self.resend_rate_limit {
-            client.set_resend_rate_limit(burst, refill_per_min);
+            client_builder = client_builder.with_resend_rate_limit(burst, refill_per_min);
         }
+        client_builder = match alloc_meter {
+            Some(meter) => client_builder.with_alloc_meter(meter),
+            None => match task_instrument.clone() {
+                Some(instrument) => client_builder.with_task_instrument(instrument),
+                None => client_builder,
+            },
+        };
+
+        let (client, sync_task_receiver) = client_builder.build().await?.into_parts();
 
         Ok(Bot {
             client,

src/client.rs

@@ -1,9 +1,11 @@
 mod accessors;
 mod adapters;
 mod app_state;
+mod builder;
 mod context_impl;
 mod device_registry;
 pub(crate) mod device_topology;
+mod extension_lifecycle;
 mod iq_ops;
 mod lid_pn;
 mod lifecycle;
@@ -13,6 +15,10 @@ pub(crate) mod offline_resume;
 mod sender_keys;
 mod sessions;
 mod voip;
+use builder::ClientAssembly;
+pub use builder::{ClientBuild, ClientBuilder, ClientBuilderError};
+use extension_lifecycle::LifecycleRegistration;
+pub use extension_lifecycle::{ClientLifecycle, ConnectionScope, ConnectionScopeState};
 pub use voip::{CallError, Voip};
 
 use crate::cache::Cache;
@@ -51,6 +57,25 @@ use portable_atomic::{AtomicI64, AtomicU64};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
 
+/// Lease that keeps raw decoded stanza events enabled for one consumer.
+///
+/// Dropping the final lease disables forwarding. The lease holds only a weak
+/// client reference, so it cannot keep the client alive.
+#[must_use = "dropping the lease immediately releases raw-node forwarding"]
+pub struct RawNodeLease {
+    client: std::sync::Weak<Client>,
+}
+
+impl Drop for RawNodeLease {
+    fn drop(&mut self) {
+        let Some(client) = self.client.upgrade() else {
+            return;
+        };
+        let previous = client.raw_node_forwarding.fetch_sub(1, Ordering::Relaxed);
+        debug_assert!(previous > 0, "raw-node forwarding lease underflow");
+    }
+}
+
 /// Filter for matching incoming stanzas (nodes) by tag and attributes.
 ///
 /// Used with [`Client::wait_for_node`] to wait for specific stanzas.
@@ -662,6 +687,8 @@ pub struct Client {
     /// error / connect_failure / disconnect. Per-connection subscribers
     /// (keepalive, request waiters, read loop, offline flush) observe this.
     pub(crate) connection_shutdown: std::sync::Mutex<wacore::runtime::ShutdownNotifier>,
+    /// Allocated only when an extension host installs lifecycle callbacks.
+    lifecycle: Option<Arc<LifecycleRegistration>>,
     /// Per-session wire I/O and activity counters. Written at the transport
     /// chokepoints (noise sender task, read loop); the keepalive dead-socket
     /// watchdog reads its activity timestamps. Snapshot via [`Client::stats`].
@@ -1012,9 +1039,8 @@ pub struct Client {
     /// its allocation-churn snapshot. Unset unless that builder method was used.
     pub(crate) alloc_meter: std::sync::OnceLock<Arc<wacore::stats::AllocMeter>>,
 
-    /// 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,
+    /// Number of consumers currently requesting `Event::RawNode` forwarding.
+    raw_node_forwarding: AtomicUsize,
 
     /// Active VoIP calls and their media-task abort handles. `abort_all` runs from the
     /// connection-cleanup path so a disconnect/reconnect tears down every in-flight call. Behind the

src/client/accessors.rs

@@ -27,17 +27,41 @@ impl Client {
         cache
     }
 
-    /// Registers an external event handler to the core event bus.
-    pub fn register_handler(&self, handler: Arc<dyn wacore::types::events::EventHandler>) {
-        self.core.event_bus.add_handler(handler);
+    /// Subscribe an external event handler with an explicit event filter.
+    pub fn subscribe(
+        &self,
+        interest: wacore::types::events::EventInterest,
+        handler: Arc<dyn wacore::types::events::EventHandler>,
+    ) -> wacore::types::events::Subscription {
+        self.core.event_bus.subscribe(interest, handler)
+    }
+
+    /// Subscribe using the handler's current registration-time interest hint.
+    pub fn subscribe_handler(
+        &self,
+        handler: Arc<dyn wacore::types::events::EventHandler>,
+    ) -> wacore::types::events::Subscription {
+        self.core.event_bus.subscribe_handler(handler)
+    }
+
+    /// Acquire raw decoded stanza forwarding for one consumer.
+    ///
+    /// `Event::RawNode` remains enabled until every acquired lease is dropped.
+    pub fn acquire_raw_node_forwarding(self: &Arc<Self>) -> RawNodeLease {
+        let incremented = self
+            .raw_node_forwarding
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |count| {
+                count.checked_add(1)
+            })
+            .is_ok();
+        assert!(incremented, "raw-node forwarding lease counter overflow");
+        RawNodeLease {
+            client: Arc::downgrade(self),
+        }
     }
 
-    /// 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);
+    pub(crate) fn raw_node_forwarding_enabled(&self) -> bool {
+        self.raw_node_forwarding.load(Ordering::Relaxed) != 0
     }
 
     /// Enable or disable skipping of history sync notifications at runtime.
@@ -516,6 +540,24 @@ impl Client {
     }
 }
 
+#[cfg(test)]
+mod raw_node_tests {
+    #[tokio::test]
+    async fn raw_node_forwarding_stays_enabled_until_the_last_lease_drops() {
+        let client = crate::test_utils::create_test_client().await;
+        assert!(!client.raw_node_forwarding_enabled());
+
+        let first = client.acquire_raw_node_forwarding();
+        let second = client.acquire_raw_node_forwarding();
+        assert!(client.raw_node_forwarding_enabled());
+
+        drop(first);
+        assert!(client.raw_node_forwarding_enabled());
+        drop(second);
+        assert!(!client.raw_node_forwarding_enabled());
+    }
+}
+
 #[cfg(test)]
 mod send_checks {
     fn assert_send<T: Send>(_: &T) {}