feat(client): let a consumer take a stanza before the built-in pipeline

src/client.rs

@@ -8,6 +8,7 @@ mod device_registry;
 pub(crate) mod device_topology;
 #[cfg(feature = "client-lifecycle")]
 mod extension_lifecycle;
+pub mod interceptor;
 mod iq_ops;
 mod lid_pn;
 mod lifecycle;
@@ -1413,6 +1414,15 @@ pub struct Client {
     /// Number of consumers currently requesting `Event::RawNode` forwarding.
     raw_node_forwarding: AtomicUsize,
 
+    /// Stanza interceptors, behind the same copy-on-write snapshot the event
+    /// bus uses: reading one costs a refcount bump, so the read loop allocates
+    /// nothing per stanza. Registering is the rare side, and pays the copy.
+    stanza_interceptors: std::sync::RwLock<Arc<Vec<interceptor::Registration>>>,
+    /// Kept alongside so the read loop can skip the lock entirely while none
+    /// are registered.
+    stanza_interceptor_count: AtomicUsize,
+    next_interceptor_id: AtomicU64,
+
     /// 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
     /// `voip` feature: it is populated only by the `voip` media facade.

src/client/accessors.rs

@@ -1,6 +1,7 @@
 //! Small accessors, config setters, node waiters and sync-error helpers.
 
 use super::*;
+use crate::client::interceptor::{InterceptorHandle, Registration, StanzaInterceptor};
 
 /// Identity for span/error tagging. Named fields, not a tuple — LID/PN transposition would
 /// otherwise be a silent, unchecked bug at call sites.
@@ -60,6 +61,77 @@ impl Client {
         self.raw_node_forwarding.load(Ordering::Relaxed) != 0
     }
 
+    /// Register an interceptor that sees each decoded stanza before the
+    /// built-in pipeline, and may take it.
+    ///
+    /// Interceptors run in registration order; the first to return
+    /// [`Interception::Handled`] wins and the rest are skipped.
+    ///
+    /// See [`crate::client::interceptor`] for what this is for and what it
+    /// costs.
+    ///
+    /// [`Interception::Handled`]: crate::client::interceptor::Interception::Handled
+    pub fn add_stanza_interceptor(
+        self: &Arc<Self>,
+        interceptor: Arc<dyn StanzaInterceptor>,
+    ) -> InterceptorHandle {
+        let id = self.next_interceptor_id.fetch_add(1, Ordering::Relaxed);
+        self.update_stanza_interceptors(|registered| {
+            registered.push(Registration { id, interceptor });
+        });
+        InterceptorHandle {
+            client: Arc::downgrade(self),
+            id,
+        }
+    }
+
+    pub(crate) fn remove_stanza_interceptor(&self, id: u64) {
+        self.update_stanza_interceptors(|registered| {
+            registered.retain(|entry| entry.id != id);
+        });
+    }
+
+    /// Whether any interceptor is registered.
+    ///
+    /// One relaxed load, so the read loop pays nothing while none are. Relaxed
+    /// is enough because the lock behind it does the synchronising: a reader
+    /// racing a registration either sees the count in time or does not, and a
+    /// stanza that arrived before the registration finished was never that
+    /// interceptor's to see.
+    pub(crate) fn has_stanza_interceptors(&self) -> bool {
+        self.stanza_interceptor_count.load(Ordering::Relaxed) != 0
+    }
+
+    /// The current interceptors.
+    ///
+    /// A refcount bump, not a copy — and the snapshot is released before any
+    /// interceptor runs, so one that registers another cannot deadlock.
+    pub(crate) fn stanza_interceptors(&self) -> Arc<Vec<Registration>> {
+        Arc::clone(
+            &self
+                .stanza_interceptors
+                .read()
+                .unwrap_or_else(std::sync::PoisonError::into_inner),
+        )
+    }
+
+    /// Copy-on-write, as the event bus does it: a reader keeps whatever
+    /// snapshot it took, so registering never blocks the read loop for longer
+    /// than the swap.
+    fn update_stanza_interceptors(&self, edit: impl FnOnce(&mut Vec<Registration>)) {
+        let mut guard = self
+            .stanza_interceptors
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut next = guard.as_ref().clone();
+        edit(&mut next);
+        // Stored while the write lock is held, so a reader never sees a count
+        // promising more than the snapshot holds.
+        self.stanza_interceptor_count
+            .store(next.len(), Ordering::Relaxed);
+        *guard = Arc::new(next);
+    }
+
     /// Enable or disable skipping of history sync notifications at runtime.
     ///
     /// When enabled, the client will acknowledge incoming history sync

src/client/interceptor.rs

@@ -0,0 +1,210 @@
+//! Taking over a stanza before the built-in pipeline sees it.
+//!
+//! The client models the stanzas it knows about and nacks the rest, which is
+//! the right default: a `<nack>` tells the server this client cannot act on
+//! something, and silence would leave the stanza in the offline queue forever.
+//!
+//! But it leaves no room for a consumer that *can* act on it. A stanza this
+//! version does not model gets nacked whether or not the application would have
+//! known what to do with it, and there is no way to say otherwise:
+//! [`StanzaRouter::register`] panics on a duplicate tag, so even a handler for
+//! an existing tag cannot be replaced.
+//!
+//! An interceptor is that room. It runs where dispatch would have, and either
+//! steps aside or claims the stanza — in which case the built-in handler is
+//! skipped and the stanza is acknowledged, so the server does not redeliver it.
+//!
+//! # What an interceptor sees
+//!
+//! Stanzas that would have reached dispatch. A response the client already
+//! correlated to a pending request, a `<xmlstreamend>` that ends the stream,
+//! and the connection-critical tags below never get there, so an interceptor
+//! does not see them either.
+//!
+//! To observe *everything* decoded, including those, use
+//! [`Event::RawNode`] — it is emitted before any of the early returns.
+//! The two are different tools: one watches, this one takes over.
+//!
+//! [`Event::RawNode`]: wacore::types::events::Event::RawNode
+//!
+//! # What cannot be claimed
+//!
+//! `success`, `failure`, `stream:error` and `ack` settle connection state:
+//! authentication, shutdown and reconnection, and the waiters a send blocks on.
+//! An interceptor that took one would not extend the client — it would leave it
+//! authenticated-but-unaware, or never reconnecting, or waiting forever on a
+//! send that already completed. They are never offered.
+//!
+//! Housekeeping is likewise untouched. Offline-sync tracking and
+//! response-waiter resolution run before dispatch and keep running whether or
+//! not a stanza is claimed.
+//!
+//! An `<iq>` the client answers on its own — a ping, a pairing step — *is*
+//! offered, because most `<iq>` traffic is exactly what a consumer would want
+//! to extend. Claiming one leaves the server without the reply it expects, so
+//! match narrowly.
+//!
+//! # Acknowledgement
+//!
+//! A claim does not change what the server is owed. Where the client would have
+//! acked, it still acks; where it would have nacked a tag it does not model,
+//! the claim turns that into an ack, because someone did handle it — answering
+//! nothing would leave the stanza in the offline queue and keep the stream
+//! recycling. Both need `id` and `from`: without them there is nothing to
+//! address, and the client would not have answered either.
+//!
+//! A tag the client *does* model but answers some other way is answered by
+//! nobody once claimed. A direct `<message>` draws a delivery `<receipt>` and
+//! an `<iq>` draws an `<iq type="result">`; a generic `<ack>` is neither, so
+//! the client does not send one. Claiming those means owing the reply.
+//!
+//! `<receipt>`, `<notification>` and `<call>` are not in that group: the client
+//! answers those with a transport `<ack>` already, so a claim leaves the ack
+//! exactly where it was.
+//!
+//! [`StanzaRouter::register`]: crate::handlers::router::StanzaRouter::register
+//!
+//! # Cost
+//!
+//! Nothing runs while no interceptor is registered: the read loop checks one
+//! relaxed atomic and carries on. Registering is what turns the check into a
+//! walk.
+//!
+//! # Example
+//!
+//! Handling a stanza the client does not model, instead of nacking it:
+//!
+//! ```no_run
+//! use std::sync::Arc;
+//! use whatsapp_rust::client::interceptor::{Interception, StanzaInterceptor};
+//! use wacore_binary::node::OwnedNodeRef;
+//!
+//! struct Vendor;
+//!
+//! impl StanzaInterceptor for Vendor {
+//!     fn intercept(&self, node: &OwnedNodeRef) -> Interception {
+//!         if node.tag() == "vendor:thing" {
+//!             // … act on it …
+//!             Interception::Handled
+//!         } else {
+//!             Interception::Pass
+//!         }
+//!     }
+//! }
+//!
+//! # fn example(client: &Arc<whatsapp_rust::Client>) {
+//! let handle = client.add_stanza_interceptor(Arc::new(Vendor));
+//! // Dropping `handle` removes it.
+//! # let _ = handle;
+//! # }
+//! ```
+
+use std::sync::{Arc, Weak};
+
+use wacore::sync_marker::MaybeSendSync;
+use wacore_binary::node::OwnedNodeRef;
+
+use crate::Client;
+
+/// What an interceptor decided about a stanza.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
+pub enum Interception {
+    /// Leave the stanza to the client.
+    ///
+    /// The default, so an interceptor that only cares about one tag needs no
+    /// branch for the rest.
+    #[default]
+    Pass,
+    /// The interceptor took the stanza.
+    ///
+    /// The built-in pipeline is skipped. The stanza is still acknowledged the
+    /// way it would have been, because the server is owed an ack either way —
+    /// withholding one leaves the stanza queued for redelivery.
+    Handled,
+}
+
+impl Interception {
+    /// Whether the built-in pipeline should be skipped.
+    #[must_use]
+    pub const fn is_handled(self) -> bool {
+        matches!(self, Self::Handled)
+    }
+}
+
+/// Sees a stanza on its way to the built-in pipeline.
+///
+/// Not every decoded stanza: see the module documentation for what never
+/// reaches this point.
+///
+/// Runs on the read loop, so it must return quickly: time spent here is time
+/// the next stanza waits. Work that can take a while belongs on a task.
+///
+/// Must not panic. Like [`EventHandler`], this is called directly by the read
+/// loop, and an unwind there takes the connection with it. The plugin host
+/// catches panics from plugins; a directly registered interceptor is trusted.
+///
+/// [`EventHandler`]: wacore::types::events::EventHandler
+pub trait StanzaInterceptor: MaybeSendSync + 'static {
+    /// Decide what happens to `node`.
+    fn intercept(&self, node: &OwnedNodeRef) -> Interception;
+}
+
+impl<F> StanzaInterceptor for F
+where
+    F: Fn(&OwnedNodeRef) -> Interception + MaybeSendSync + 'static,
+{
+    fn intercept(&self, node: &OwnedNodeRef) -> Interception {
+        self(node)
+    }
+}
+
+/// Keeps an interceptor registered. Dropping it removes the interceptor.
+///
+/// Holds a weak client reference, so a forgotten handle cannot keep a client
+/// alive.
+#[must_use = "dropping the handle immediately removes the interceptor"]
+#[derive(Debug)]
+pub struct InterceptorHandle {
+    pub(crate) client: Weak<Client>,
+    pub(crate) id: u64,
+}
+
+impl Drop for InterceptorHandle {
+    fn drop(&mut self) {
+        if let Some(client) = self.client.upgrade() {
+            client.remove_stanza_interceptor(self.id);
+        }
+    }
+}
+
+/// One registered interceptor.
+#[derive(Clone)]
+pub(crate) struct Registration {
+    pub(crate) id: u64,
+    pub(crate) interceptor: Arc<dyn StanzaInterceptor>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn pass_is_the_default_so_narrow_interceptors_stay_short() {
+        assert_eq!(Interception::default(), Interception::Pass);
+        assert!(!Interception::Pass.is_handled());
+        assert!(Interception::Handled.is_handled());
+    }
+
+    #[test]
+    fn a_closure_is_an_interceptor() {
+        fn takes(_: impl StanzaInterceptor) {}
+        takes(|_node: &OwnedNodeRef| Interception::Pass);
+    }
+
+    #[test]
+    fn interception_is_comparable_and_debuggable() {
+        assert_eq!(Interception::Pass, Interception::Pass);
+        assert_ne!(Interception::Pass, Interception::Handled);
+        assert!(!format!("{:?}", Interception::Handled).is_empty());
+    }
+}

src/client/lifecycle.rs

@@ -506,6 +506,9 @@ impl Client {
             saver_handle: std::sync::OnceLock::new(),
             alloc_meter: std::sync::OnceLock::new(),
             raw_node_forwarding: AtomicUsize::new(0),
+            stanza_interceptors: std::sync::RwLock::new(Arc::new(Vec::new())),
+            stanza_interceptor_count: AtomicUsize::new(0),
+            next_interceptor_id: AtomicU64::new(0),
             #[cfg(feature = "voip-runtime")]
             call_registry: Arc::new(wacore::voip::CallRegistry::new()),
             #[cfg(feature = "voip-runtime")]