diff --git a/advanced/plugins.mdx b/advanced/plugins.mdx index 030b4c70..03df4b44 100644 --- a/advanced/plugins.mdx +++ b/advanced/plugins.mdx @@ -10,7 +10,7 @@ whatsapp-rust supports **native plugins**: build-time, type-safe extensions that This is a low-level extension seam. If you're writing an application that just sends and receives messages, use [`Bot`](/api/bot) directly — you don't need this page. Reach for native plugins when you're building a **reusable component** (metrics, a search index, a moderation layer) that should install once, observe events non-blockingly, own its own background tasks across reconnects, and expose a typed API to the rest of your application. -The plugin host is intentionally native-only today: it is trusted in-process Rust code, not a sandboxed or dynamically-loaded extension model. There is no dynamic (`.so`/`.wasm`) loading, no ingress interception of inbound stanzas, and no pre-ack decision hook. +The plugin host is intentionally native-only today: it is trusted in-process Rust code, not a sandboxed or dynamically-loaded extension model. There is no dynamic (`.so`/`.wasm`) loading and no pre-ack decision hook. Ingress interception — claiming a stanza before the built-in pipeline — is supported via the `PluginCapability::StanzaInterception` capability below. ## Enabling the feature @@ -107,6 +107,7 @@ A plugin's manifest requests capabilities, and `PluginContext` only exposes the | `PluginCapability::Messaging` | `PluginMessaging` | High-level message sends | | `PluginCapability::Iq` | `PluginIq` | Typed `IqSpec` execution | | `PluginCapability::PluginEvents` | `PluginEvents` | Publishing custom events, scoped to the plugin's own namespace | +| `PluginCapability::StanzaInterception` | `PluginStanzaInterception` | Claiming a decoded stanza before the built-in pipeline sees it | This is API shaping, not a sandbox: a native plugin is trusted in-process Rust code and can use any dependency available to your build. It exists so a plugin's `install` signature documents what it actually touches, and so `PluginContext` never hands out the raw backend or Signal stores. @@ -145,7 +146,7 @@ The other capabilities follow the same pattern: `context.tasks()`, `context.mess Capability handles hold a weak reference to the client internally and reject calls once the client has shut down, so a plugin API that outlives the client (for example, one your application keeps an `Arc` to) fails safely instead of resurrecting it. -`PluginCoreEvents::subscribe` returns an RAII token for the registration. Dropping it, or explicitly unsubscribing, removes the handler immediately. +`PluginCoreEvents::subscribe` returns an RAII token for the registration. Dropping it, or explicitly unsubscribing, removes the handler immediately. `PluginStanzaInterception::register` (below) returns the same shape of token. Both are indexed weakly in the host, so terminal shutdown can invalidate a token a plugin API retained without extending the lifetime of one the plugin already released. ## Lifecycle and task scopes @@ -178,9 +179,53 @@ Each delivered envelope carries six fields: the plugin ID, the topic, a schema v `PluginEventSubscription` is an RAII endpoint — dropping it removes all of that subscriber's selectors from the router at once. +## Stanza interception + +`PluginCapability::CoreEvents` lets a plugin *watch* the stream. `PluginCapability::StanzaInterception` lets it *act*: a plugin that models a stanza this client does not can claim it before the built-in pipeline sees it, instead of watching it get nacked. The underlying seam — `Client::add_stanza_interceptor`, the `StanzaInterceptor` trait, what is never offered, and the ack a claim still owes the server — is documented on [`add_stanza_interceptor`](/api/client#add_stanza_interceptor); this section covers what the plugin host adds on top of it. + +```rust +use std::sync::Arc; + +use whatsapp_rust::{PluginCapability, PluginStanzaInterception}; +use whatsapp_rust::client::interceptor::{Interception, StanzaInterceptor}; + +impl ClientPlugin for VendorPlugin { + type Api = VendorApi; + + fn manifest(&self) -> PluginManifest { + PluginManifest::new("example.vendor", "0.1.0") + .with_capability(PluginCapability::StanzaInterception) + } + + fn install(&self, context: PluginContext) -> PluginFuture<'_, Result>> { + Box::pin(async move { + let interception = context + .stanza_interception() + .cloned() + .expect("StanzaInterception capability was requested in the manifest above"); + let registration = interception.register(Arc::new(VendorInterceptor))?; + Ok(Arc::new(VendorApi { registration })) + }) + } +} +``` + +`PluginStanzaInterception::register` returns a `PluginInterceptorRegistration` — the same ownership-token shape `PluginCoreEvents::subscribe` returns. Dropping it unregisters; `unregister()` is the explicit form; the host indexes it weakly so terminal shutdown invalidates a token a plugin API retained. + +The host adds two things a directly registered interceptor doesn't get: + +- **Panic isolation.** `StanzaInterceptor` asks implementations not to panic, and a directly registered interceptor is trusted to honor that — the read loop calls it inline, so an unwind there takes the connection with it. A plugin is not trusted with that: one faulty plugin must not kill the connection. A panicking interceptor is counted (`stanza_interception_panics` in `PluginStats`), logged, and treated as `Interception::Pass` — the outcome of the plugin not being there — which also degrades the plugin's `PluginHealth`. +- **Terminal invalidation.** Like every other plugin registration, a stanza interceptor is closed on shutdown, so a client that is shutting down is not still asking a plugin what to do with its stanzas. + + +Interception runs before the built-in pipeline, so a claimed stanza is one the client did no Signal work on at all: nothing decrypted, no session mutated, no prekey consumed. There is no half-advanced state to reconcile — the property that makes claiming safe to reason about without a new durability contract. But it also means the claim is final: the ack that follows tells the server not to redeliver, so a claimed `` stays undecrypted forever and a claimed prekey-bearing `` never tops up prekeys. A plugin claiming stanzas that carry Signal state takes over that responsibility whole — match narrowly, and see [Signal durability](/advanced/signal-protocol) for what the responsibility involves. + + +`Client::memory_report()`'s [`plugin_stanza_interceptors`](/api/client#memory_report) field folds in every plugin's active interceptor count. + ## Diagnostics -`Client::plugin_stats()` returns a `PluginStats` snapshot per installed plugin: lifecycle state (`PluginState`), sticky health (`PluginHealth`), callback failures, spawned-task panics, drain timeouts, active task scopes, and subscription/publisher counters. `PluginEventRouter::stats()` and `PluginEvents::stats()` report queue depth and backpressure totals for custom events. `Client::memory_report()` folds in plugin resource accounting. +`Client::plugin_stats()` returns a `PluginStats` snapshot per installed plugin: lifecycle state (`PluginState`), sticky health (`PluginHealth`), callback failures, spawned-task panics, drain timeouts, active task scopes, subscription/publisher counters, and — since stanza interception — `stanza_interceptors` and `stanza_interception_panics`. `PluginEventRouter::stats()` and `PluginEvents::stats()` report queue depth and backpressure totals for custom events. `Client::memory_report()` folds in plugin resource accounting. These snapshots are on-demand and approximate under concurrency, and — matching the rest of the client's [observability](/advanced/observability) surface — never include JIDs, phone numbers, or message bodies. @@ -190,7 +235,7 @@ The workspace ships `plugins/metrics` as a public-API conformance example: a met ## What's not supported yet -The current host deliberately does not support dynamic plugin loading, ingress interception or pre-ack decisions on inbound stanzas, a foreign (non-Rust) wire protocol, or process isolation/sandboxing. These are real design commitments, not just missing features, and would only be added behind a concrete consumer need. +The current host deliberately does not support dynamic plugin loading, pre-ack decisions on inbound stanzas, a foreign (non-Rust) wire protocol, or process isolation/sandboxing. These are real design commitments, not just missing features, and would only be added behind a concrete consumer need. Ingress interception (claiming a stanza, above) shipped once its interaction with Signal durability had a design — see the warning in [Stanza interception](#stanza-interception). ## See also diff --git a/api/client.mdx b/api/client.mdx index a630c5b9..1f9881bd 100644 --- a/api/client.mdx +++ b/api/client.mdx @@ -1846,6 +1846,123 @@ bot.on_event(|event, _client| async move { Only enable this when you need raw protocol access. Every decoded stanza triggers the event, which adds overhead to the message processing pipeline. +### add_stanza_interceptor + +```rust +pub fn add_stanza_interceptor( + self: &Arc, + interceptor: Arc, +) -> InterceptorHandle +``` + +Register an interceptor that sees each decoded stanza before the built-in pipeline, and may take it. This is the seam for acting on a stanza this client does not model — `StanzaRouter::register` panics on a duplicate tag, so even an existing tag can't be handled differently any other way — instead of watching it get nacked. + + + The interceptor to register. A plain closure of type `Fn(&OwnedNodeRef) -> Interception` implements `StanzaInterceptor` too, so `client.add_stanza_interceptor(Arc::new(|node: &OwnedNodeRef| { .. }))` works without a named type. + + + + RAII token for the registration. Dropping it removes the interceptor. The handle holds only a weak client reference, so a forgotten handle cannot keep the client alive, and dropping one after its client is already gone is a no-op rather than a panic. + + +```rust +use wacore::sync_marker::MaybeSendSync; + +pub trait StanzaInterceptor: MaybeSendSync + 'static { + fn intercept(&self, node: &OwnedNodeRef) -> Interception; +} + +pub enum Interception { + /// Leave the stanza to the client. The default. + Pass, + /// The interceptor took the stanza; the built-in pipeline is skipped. + Handled, +} +``` + +`MaybeSendSync` is `Send + Sync` on native targets and carries no bounds on `wasm32`, matching the convention used by `EventHandler`, `Transport`, and `HttpClient`. + +Interceptors run in registration order; the first one to return `Interception::Handled` wins and the rest — including the built-in pipeline — are skipped. Registration order is therefore priority order: an earlier registration can shadow a later one. + +**What an interceptor never sees:** `success`, `failure`, `stream:error`, and `ack` settle connection state (authentication, shutdown/reconnection, and the waiters a send blocks on), and a server-initiated `` ping is withheld for the same reason — a claimed ping is a pong never sent, and the server drops the connection over it. Offline-sync tracking and response-waiter resolution run before dispatch and keep running whether or not a stanza is claimed. Every other stanza, including `` traffic the client already answers on its own, is offered. + +**What claiming owes the server:** a claim doesn't change what the server is owed. Where the client would have acked a stanza, it still acks; where it would have nacked a tag it doesn't model, the claim turns that into an ack instead, since something did handle it — answering nothing would leave the stanza in the offline queue. A tag the client models but answers some other way (a delivery `` for a direct ``, an ``) gets nothing from the claimed-stanza path — the claimant owes that reply itself. + + +Cost while unused: one relaxed atomic load on the read loop, checked before any lock. Registering is what turns the check into a walk over the registered interceptors. + + +**Example:** +```rust +use whatsapp_rust::client::interceptor::{Interception, StanzaInterceptor}; +use wacore_binary::node::OwnedNodeRef; +use std::sync::Arc; + +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 + } + } +} + +let handle = client.add_stanza_interceptor(Arc::new(Vendor)); +// Dropping `handle` removes it. +``` + + +Interception runs before the built-in pipeline — including Signal decryption — so a claimed `` was never decrypted and a claimed prekey-bearing `` never tops up prekeys. The ack that follows tells the server not to redeliver, so that work never happens again. Match narrowly. + + +See the `whatsapp_rust::client::interceptor` module for the full contract, and [Native plugins](/advanced/plugins#stanza-interception) for the capability-gated version available to plugins. + +### acquire_decrypted_payload_forwarding + +```rust +pub fn acquire_decrypted_payload_forwarding(self: &Arc) -> DecryptedPayloadLease +``` + +Acquire a lease that keeps [`Event::DecryptedPayload`](/concepts/events#decryptedpayload) enabled for one consumer. The lease is necessary but not sufficient: a handler that has narrowed its `interest()` away from the default `EventInterest::ALL` also needs `EventKind::DecryptedPayload` added back in, or it won't see the event even while a lease is held. The event carries a message's plaintext *before* it is decoded into a `wa::Message` — the only way to recover a payload that decrypts successfully but fails to decode (a field a build predates, a message type it doesn't model). Nothing can ask for those bytes again: opening them already consumed state that won't recur — the Signal ratchet advances, or (for a bot's `message_secret` payload) the single-use secret is spent — so the same ciphertext will never open a second time. + + + RAII lease. `Event::DecryptedPayload` stays enabled until every acquired lease is dropped — hold it for as long as you want the event forwarded. The lease holds only a weak client reference, so it cannot keep the client alive. + + + +While no lease is held, nothing is emitted and nothing is cloned — the path costs one relaxed atomic load. Under a lease, the forwarded payload is the same `bytes::Bytes` the decoder receives, so forwarding it is a refcount bump rather than a copy. + + +**Example:** +```rust +use wacore::types::events::{ChannelEventHandler, Event}; + +let _lease = client.acquire_decrypted_payload_forwarding(); + +let (handler, event_rx) = ChannelEventHandler::new(); +client.register_handler(handler); + +while let Ok(event) = event_rx.recv().await { + if let Event::DecryptedPayload(payload) = &*event { + // `payload.payload` is the unpadded plaintext, exactly as decoding + // will receive it — record it, or inspect it when decoding later + // fails for this message. + println!( + "enc #{} ({}) for {}: {} bytes", + payload.enc_index, + payload.enc_type, + payload.info.id, + payload.payload.len(), + ); + } +} +// Drop `_lease` to stop forwarding. +``` + --- ## Call management @@ -2104,6 +2221,8 @@ Entry counts plus estimated retained heap bytes for the client's internal collec | `history_sync_payload_bytes_peak` | `u64` | Lifetime high-water mark of logical compressed-payload bytes | | `chatstate_handlers` | `usize` | Registered chat state handlers | | `custom_enc_handlers` | `usize` | Registered custom encryption handlers | +| `stanza_interceptors` | `usize` | Registered stanza interceptors ([`add_stanza_interceptor`](#add_stanza_interceptor)). A handle that outlives its interest leaves one registered, and a leak here costs a walk on every stanza — which this count is what makes visible ([#1239](https://github.com/oxidezap/whatsapp-rust/pull/1239)) | +| `plugin_stanza_interceptors` | `u64` | Behind the `plugins` feature: sum of every installed plugin's active [stanza interceptors](/advanced/plugins#stanza-interception) ([#1241](https://github.com/oxidezap/whatsapp-rust/pull/1241)) | `CollectionStats` carries both `entries: u64` and `bytes: u64`. `MemoryReport::total_estimated_bytes(&self) -> u64` sums `.bytes` across every byte-carrying field. `MemoryReport` implements `Display` for a pretty-printed, human-readable breakdown. This output includes an `--- In-flight history sync ---` section with the two peak fields above. diff --git a/concepts/events.mdx b/concepts/events.mdx index f0fba400..9779e8e3 100644 --- a/concepts/events.mdx +++ b/concepts/events.mdx @@ -212,6 +212,9 @@ pub enum Event { PairPasskeyRequest(PairPasskeyRequest), PairPasskeyConfirmation(PairPasskeyConfirmation), PairPasskeyError(PairPasskeyError), + + // Decrypted payload (opt-in) + DecryptedPayload(DecryptedPayload), } ``` @@ -2645,6 +2648,56 @@ Event::RawNode(node) => { `RawNode` is skipped during serialization (`#[serde(skip)]`). Enable it only when debugging or building protocol-level tooling, as it dispatches for every incoming stanza. +## Decrypted payload events + +### `DecryptedPayload` + +**Emitted:** One decrypted `` payload, after unpadding and *before* it is decoded into a `wa::Message`. To receive this event, hold a lease from `client.acquire_decrypted_payload_forwarding()` and include `EventKind::DecryptedPayload` in your handler's `interest()`. While no lease is held, nothing is emitted and nothing is cloned. + +```rust +#[derive(Debug, Clone, Serialize, bon::Builder)] +#[non_exhaustive] +pub struct DecryptedPayload { + pub info: Arc, + pub enc_index: usize, + pub enc_type: &'static str, + #[serde(skip)] + pub payload: Bytes, +} + +Event::DecryptedPayload(DecryptedPayload) +``` + +**Fields:** +- `info` — Which message this came from. +- `enc_index` — Which `` of the stanza produced these bytes, counting from zero in the order the client enumerates them: the stanza's direct `` children first, then the ones under `` addressed to this device (the fan-out shape, where one stanza carries a copy per device and only yours is yours to decrypt). This is a position in that concatenation, not a child index or a position within `enc_type`'s bucket — an `` that produces no payload still consumes its slot, so a consumer correlating a forwarded payload back to its node has to walk the stanza the same way. +- `enc_type` — The `type` attribute the `` carried: `msg`, `pkmsg`, `skmsg`, … +- `payload` — The plaintext, unpadded, exactly as decoding receives it. A `Bytes`, so forwarding it is a refcount bump, not a copy. + +This is a library extension with no WhatsApp Web equivalent. It exists because a plaintext that decrypts but fails to decode is otherwise lost: `handle_decrypted_plaintext` turns bytes into `wa::Message`, and when that decode fails — a field a build predates, a message type it doesn't model — the bytes disappear. Nothing can ask for them again, because opening them already consumed state that won't recur: the Signal ratchet advances, so the same ciphertext will never decrypt a second time. `DecryptedPayload` fires whether or not the decode that follows succeeds, which is the point: the failing case is the one with nothing else to look at. It also enables recording traffic for faithful replay (re-encoding a decoded `Message` does not reproduce the original bytes) and decoding with a newer protobuf than the running build carries. + +It's also emitted on the bot-message-secret path (`msg_secret.rs`), ahead of the same decode, where the secret a `message_secret` payload was opened with is single-use rather than ratchet-advanced — the same "cannot be asked for again" property, for a different reason. + +**Example:** +```rust +let _lease = client.acquire_decrypted_payload_forwarding(); + +// In your event handler: +Event::DecryptedPayload(payload) => { + println!( + "enc #{} ({}) for {}: {} bytes", + payload.enc_index, payload.enc_type, payload.info.id, payload.payload.len(), + ); +} +// Drop `_lease` to stop forwarding. +``` + + +`payload` is skipped during serialization (`#[serde(skip)]`) — like `RawNode`, `Serialize` on an event is for diagnostics, and no text format carries raw bytes without an encoding choice this type has no business making. + + +See [`acquire_decrypted_payload_forwarding`](/api/client#acquire_decrypted_payload_forwarding) for the lease API. + ## Event handler patterns ### Bot builder pattern