diff --git a/advanced/inbound-durability.mdx b/advanced/inbound-durability.mdx index 7ef04398..950a1c10 100644 --- a/advanced/inbound-durability.mdx +++ b/advanced/inbound-durability.mdx @@ -1,6 +1,6 @@ --- title: Inbound Durability Hook -description: Opt in to at-least-once message delivery by deferring the transport ack until your consumer durably commits each message. +description: Opt in to at-least-once message delivery by deferring the transport ack until your consumer durably commits each batch of messages. --- ## Overview @@ -9,9 +9,9 @@ By default the client acknowledges a message to the WhatsApp server **as soon as Registering an `InboundDurabilityHook` converts the consumer to **at-least-once delivery**: -1. The decrypted message is buffered durably in the `pending_inbound_messages` table **before** the Signal ratchet is flushed. -2. Your hook is awaited. On `Ok` the message is acked and the buffer row is cleared. -3. On `Err` (or a crash), the ack is suppressed. The server redelivers the message on the next connect, where the hook runs again from the buffered copy. +1. The decrypted message(s) are buffered durably in the `pending_inbound_messages` table **before** the Signal ratchet is flushed. +2. Your hook is awaited with the whole batch. On `Ok` every message in the batch is acked and its buffer row cleared. +3. On `Err` (or a crash), all their acks are suppressed. The server redelivers the batch on the next connect, where the hook runs again from the buffered copies. Default behavior is unchanged — with no hook registered nothing is buffered and the ack path is identical to before. @@ -19,6 +19,23 @@ Default behavior is unchanged — with no hook registered nothing is buffered an This is the same gap whatsmeow closes with `SynchronousAck` + `EnableDecryptedEventBuffer`. The design follows whatsmeow's decrypt-buffer approach rather than a global gate. +## Batching + +Live traffic is delivered to the hook one message at a time — a batch of one, committed immediately, so latency is unchanged from the previous per-message behavior. + +During the **offline drain** (the backlog replayed on reconnect), the client accumulates decrypted messages and commits them as a batch, mirroring WhatsApp Web's `MessageProcessorCache` granularity. A batch flushes on whichever trigger fires first: + +| Trigger | Value | +|---|---| +| Message count | 200 | +| Encoded size | 4 MiB | +| Timeout since first buffered message | 3 seconds | +| End of drain / disconnect / reconnect | forced flush | + +These triggers are internal constants, not currently exposed as configuration. The drain→live transition is raceless: the tail batch of the drain always commits before any live-mode message is processed, so a consumer never observes drain and live messages out of order across the boundary. + +Within a batch, the commit order is: durable buffer write (one transaction) → Signal-cache flush → your hook → buffer clear → acks → `Event::Messages` dispatch. A failure at any step leaves the **entire batch** unacked, and the server redelivers all of it — so your hook must commit a batch all-or-nothing. + ## Opting In ```rust @@ -30,16 +47,21 @@ struct MyStore { /* your DB connection */ } #[async_trait] impl InboundDurabilityHook for MyStore { - async fn on_message( + async fn on_messages( &self, _client: Arc, - info: &MessageInfo, - message: &wa::Message, + batch: &[InboundMessage], ) -> anyhow::Result<()> { - // Durably commit the message (INSERT, enqueue to broker, etc.). - // Return Ok only after the commit is durable. - // Return Err to suppress the ack and trigger redelivery. - my_db_insert(&info.id, message).await?; + // Ideally a single INSERT/transaction over the whole batch. This + // loop commits per item instead, so a failure partway through (the + // `?` on a later item) leaves earlier items already durably + // committed — yet the SDK still suppresses every ack and redelivers + // the whole batch on Err. `my_db_insert` MUST be an idempotent + // upsert (e.g. `INSERT ... ON CONFLICT DO NOTHING`) so that replay + // of an already-committed item is a no-op, not a duplicate. + for item in batch { + my_db_insert(&item.info.id, &item.message).await?; + } Ok(()) } } @@ -49,7 +71,8 @@ let bot = Bot::builder() // Opt in — without this call the client keeps its default at-most-once behavior. .with_inbound_durability_hook(MyStore { /* ... */ }) .on_message(|ctx| async move { - // Event handlers still fire as normal; the ack is deferred in the background. + // Event handlers still fire once per message, in arrival order; + // the ack is deferred in the background. println!("received: {}", ctx.info.id); }) .build() @@ -61,42 +84,48 @@ let bot = Bot::builder() ```rust use async_trait::async_trait; use std::sync::Arc; +use wacore::types::events::InboundMessage; #[async_trait] pub trait InboundDurabilityHook { - async fn on_message( - &self, - client: Arc, - info: &MessageInfo, - message: &wa::Message, - ) -> anyhow::Result<()>; + async fn on_messages(&self, client: Arc, batch: &[InboundMessage]) -> anyhow::Result<()>; } ``` | Parameter | Description | |-----------|-------------| -| `client` | The active client — use it for lookups, **not** for sending to the same sender (see [Caveats](#caveats)). | -| `info` | Message metadata (id, sender, chat, timestamp). | -| `message` | The decrypted message proto. | +| `client` | The active client — use it for lookups, **not** for sending to a sender present in the batch (see [Caveats](#caveats)). | +| `batch` | The decrypted messages to commit, in arrival order. A batch of one on live traffic; possibly many during the offline drain. | -Return `Ok(())` once the commit is durable. Return `Err` to suppress the ack. +Each `InboundMessage` carries: + +```rust +pub struct InboundMessage { + pub message: Arc, + pub info: Arc, +} +``` + +Return `Ok(())` once the whole batch is durably committed. Return `Err` to suppress every ack in the batch. + +`InboundMessage`, `MessageBatch`, and `BatchOrigin` are re-exported from the crate `prelude`. ## Idempotency Requirement -At-least-once means the hook **will be called more than once for the same message** when a crash occurs after the consumer commits but before the ack lands. Your hook **must be idempotent**. +At-least-once means the hook **will be called more than once for the same message** when a crash occurs after the consumer commits but before the ack lands. Your hook **must be idempotent**, and since a failed batch is redelivered whole, a partially-applied batch commit must also be safe to re-run. Deduplicate by the full triplet `(info.source.chat, info.source.sender, info.id)` — **not** `info.id` alone. Stanza IDs are only unique within a `(chat, sender)` pair, so two different chats can reuse the same ID string. ```rust -// Correct idempotency key +// Correct idempotency key, per item in the batch let key = ( - info.source.chat.to_string(), - info.source.sender.to_string(), - info.id.clone(), + item.info.source.chat.to_string(), + item.info.source.sender.to_string(), + item.info.id.clone(), ); // Wrong — id alone is not globally unique -let key = info.id.clone(); +let key = item.info.id.clone(); ``` In SQL, a `UNIQUE` constraint or `INSERT OR IGNORE` on `(chat, sender, id)` is the most robust approach. @@ -106,7 +135,7 @@ In SQL, a `UNIQUE` constraint or `INSERT OR IGNORE` on `(chat, sender, id)` is t When the server redelivers a message the client previously did not ack: 1. The client detects the duplicate stanza. -2. If a hook is registered and a buffered copy exists in `pending_inbound_messages`, the hook is run from that copy. +2. If a hook is registered and a buffered copy exists in `pending_inbound_messages`, the message re-enters the commit pipeline (it can be grouped into the same batch as other stanzas being processed at the time) and the hook runs from the buffered copy. 3. On `Ok` the buffer row is cleared and the message is acked. 4. On `Err` the buffer is kept; the hook runs again on the next redelivery. 5. If no buffered copy exists (genuine duplicate already committed), the message is acked directly without invoking the hook. @@ -115,36 +144,38 @@ A 7-day retention sweep removes rows that a permanently-failing hook would other ## Backend Requirement -Durable cross-crash replay requires a backend that implements the four pending-inbound methods on `ProtocolStore`: +Durable cross-crash replay requires a backend that implements the pending-inbound methods on `ProtocolStore`: - `store_pending_inbound` — write the decrypted message bytes before the hook runs - `get_pending_inbound` — read the buffer on redelivery - `delete_pending_inbound` — clear the buffer after the hook commits - `delete_expired_pending_inbound` — retention sweep (called unconditionally from keepalive) -The bundled `SqliteStore` implements all four. Custom backends that do not implement them return an error from the defaults, which causes `Bot::build()` to fail with `BotBuilderError::UnsupportedDurabilityBackend` — a clear error rather than a silent runtime degradation. +Two additional batch-oriented methods, `store_pending_inbound_batch` and `delete_pending_inbound_batch`, default to looping the single-row methods above, so existing custom backends keep working unchanged. The bundled `SqliteStore` overrides both to commit a whole batch in one transaction — see [Custom Backends](/guides/custom-backends#pending-inbound-buffer) if you want the same atomicity in your own backend. -If you use a custom backend and want to opt in to the durability hook, implement the four methods. See [Custom Backends](/guides/custom-backends#pending-inbound-buffer) for implementation guidance. +Backends that implement none of these return an error from the defaults, which causes `Bot::build()` to fail with `BotBuilderError::UnsupportedDurabilityBackend` — a clear error rather than a silent runtime degradation. ## Caveats -**At-least-once, not exactly-once.** A crash after your consumer commits but before the ack lands replays the message. Your hook must be idempotent (deduplicate by `(chat, sender, id)`). +**At-least-once, not exactly-once.** A crash after your consumer commits but before the ack lands replays the message (or its whole batch). Your hook must be idempotent (deduplicate by `(chat, sender, id)`). **Backpressure.** The hook is awaited inside the receive pipeline. A slow hook backpressures inbound processing for the duration of the commit — the same trade-off as whatsmeow's synchronous ack. Persist and return; spawn any reply logic after the hook returns `Ok`. -**No synchronous sends to the same sender.** During 1:1 message processing the per-sender Signal lock is held. Performing a synchronous client operation to the same sender inside the hook (e.g. a blocking reply) will deadlock. Use `tokio::spawn` if you need to reply from within the hook. +**No synchronous sends to a sender in the batch.** During 1:1 message processing the per-sender Signal lock is held. Performing a synchronous client operation to a sender present in the batch (e.g. a blocking reply) will deadlock. Use `tokio::spawn` if you need to reply from within the hook. -**Scope.** The hook covers end-to-end encrypted messages (1:1 and group). Newsletter and broadcast channel messages use a separate ack path and are not gated by the hook. +**Scope.** The hook covers end-to-end encrypted messages (1:1 and group). Newsletter and broadcast channel messages use a separate ack path and are never gated by the hook — they dispatch `Event::Messages` directly. PDO placeholder recoveries (`info.unavailable_request_id` set) bypass the hook the same way. -**Buffer-write failure.** If the durable buffer write itself fails (e.g. disk full), the ack is suppressed, but if the process does not crash the Signal ratchet still advances. That one message degrades to at-most-once on its next redelivery (it can no longer be decrypted and there is no buffered copy to replay). The guarantee holds whenever the buffer write succeeds. +**Buffer-write failure.** If the durable buffer write itself fails (e.g. disk full), the acks for that batch are suppressed, but if the process does not crash the Signal ratchet still advances. Those messages degrade to at-most-once on their next redelivery (they can no longer be decrypted and there is no buffered copy to replay). The guarantee holds whenever the buffer write succeeds. **Redelivery `info` fields.** On a redelivery replay, `info` is re-parsed from the stanza. A few fields derived during the first dispatch (the ephemeral timer, encrypted comment threading) may be absent. The `message` body is always the original. +**`Event::Messages` is at-least-once too, when a hook is registered.** A redelivery whose buffered copy survived (e.g. the post-commit cleanup failed and the ack was lost) replays through the same commit and dispatches the event again — event handlers need the same idempotency discipline as the hook if they perform side effects. + ## Full Example -The repository ships `examples/durability_hook.rs` — a file-backed archiver that appends each message to disk with `fsync` before returning `Ok`, and seeds the deduplication set from the archive on startup so dedupe survives a restart. +The repository ships `examples/durability_hook.rs` — a file-backed archiver that appends each message in a batch to disk with a single `fsync` before returning `Ok`, and seeds the deduplication set from the archive on startup so dedupe survives a restart. ```bash cargo run --example durability_hook @@ -153,41 +184,59 @@ cargo run --example durability_hook Key patterns from that example: ```rust -#[async_trait] +#[async_trait::async_trait] impl InboundDurabilityHook for InboxArchiver { - async fn on_message( + async fn on_messages( &self, _client: Arc, - info: &MessageInfo, - message: &wa::Message, + batch: &[whatsapp_rust::types::events::InboundMessage], ) -> anyhow::Result<()> { - let key = ( - info.source.chat.to_string(), - info.source.sender.to_string(), - info.id.clone(), - ); - - // Idempotency check first. - if self.seen.lock()?.contains(&key) { - return Ok(()); + // Live traffic arrives one message at a time; an offline drain hands + // over a whole batch. Either way the commit below is a single append + + // fsync, so the durability cost amortizes over the batch. + let mut lines = String::new(); + let mut keys: Vec = Vec::with_capacity(batch.len()); + { + let seen = self.seen.lock().map_err(|_| anyhow::anyhow!("seen lock poisoned"))?; + for m in batch { + let key: CommitKey = ( + m.info.source.chat.to_string(), + m.info.source.sender.to_string(), + m.info.id.clone(), + ); + // Dedup against the archive AND earlier entries of this same + // batch, so one fsync can never append a key twice. + if seen.contains(&key) || keys.contains(&key) { + continue; + } + let preview = m.message.conversation.as_deref().unwrap_or("").replace(['\t', '\n'], " "); + lines.push_str(&format!("{}\t{}\t{}\t{preview}\n", key.0, key.1, key.2)); + keys.push(key); + } } - // Durable commit on a blocking thread — disk I/O must not run - // on the async receive thread. - let file = Arc::clone(&self.file); - let line = format!("{}\t{}\t{}\n", key.0, key.1, key.2); - tokio::task::spawn_blocking(move || -> std::io::Result<()> { - let mut f = file.lock().unwrap(); - f.write_all(line.as_bytes())?; - f.sync_all() // Must sync before returning Ok - }).await??; - - self.seen.lock()?.insert(key); + if !keys.is_empty() { + // Durable commit on a blocking thread: append then fsync — all-or- + // nothing for the batch. Returning Ok only after sync_all means + // "safe to ack every message"; any error returns Err, so the acks + // are suppressed and the server redelivers the batch later. + let file = Arc::clone(&self.file); + tokio::task::spawn_blocking(move || -> std::io::Result<()> { + let mut f = file.lock().expect("file lock poisoned"); + f.write_all(lines.as_bytes())?; + f.sync_all() + }).await.map_err(|e| anyhow::anyhow!("archive write task failed: {e}"))??; + + let mut seen = self.seen.lock().map_err(|_| anyhow::anyhow!("seen lock poisoned"))?; + for key in keys { + seen.insert(key); + } + } Ok(()) } } ``` -Always use `tokio::task::spawn_blocking` for disk I/O inside the hook. Blocking calls on the async thread stall the entire receive pipeline. +Always use `tokio::task::spawn_blocking` for disk I/O inside the hook. Blocking calls on the async thread stall the entire receive pipeline, including the batch that is waiting to commit. diff --git a/api/bot.mdx b/api/bot.mdx index 7d708829..1784ab3f 100644 --- a/api/bot.mdx +++ b/api/bot.mdx @@ -25,7 +25,7 @@ The Bot is the **recommended way** to use whatsapp-rust. It provides sensible de ```rust use whatsapp_rust::bot::Bot; use whatsapp_rust::TokioRuntime; -use wacore::types::events::Event; +use wacore::types::events::{Event, InboundMessage}; let mut bot = Bot::builder() .with_backend(backend) @@ -34,8 +34,10 @@ let mut bot = Bot::builder() .with_runtime(TokioRuntime) .on_event(|event, client| async move { match &*event { - Event::Message(msg, info) => { - println!("Message from {}: {:?}", info.source.sender, msg); + Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + println!("Message from {}: {:?}", info.source.sender, msg); + } } Event::Connected(_) => { println!("Connected to WhatsApp!"); @@ -230,13 +232,15 @@ use waproto::whatsapp as wa; Bot::builder() .on_event(|event, client| async move { match &*event { - Event::Message(msg, info) => { + Event::Messages(batch) => { // Reply to messages - let reply = wa::Message { - conversation: Some("Hello back!".to_string()), - ..Default::default() - }; - let _ = client.send_message(info.source.chat.clone(), reply).await; + for InboundMessage { info, .. } in batch.iter() { + let reply = wa::Message { + conversation: Some("Hello back!".to_string()), + ..Default::default() + }; + let _ = client.send_message(info.source.chat.clone(), reply).await; + } } Event::Connected(_) => { println!("Bot online!"); @@ -264,9 +268,9 @@ Like `on_event`, but registers the handler with a narrowed [`EventInterest`](/co use wacore::types::events::EventKind; Bot::builder() - .on_event_for(&[EventKind::Message, EventKind::Connected], |event, client| async move { + .on_event_for(&[EventKind::Messages, EventKind::Connected], |event, client| async move { match &*event { - Event::Message(msg, info) => { /* … */ } + Event::Messages(batch) => { /* … */ } Event::Connected(_) => println!("online"), _ => {} } @@ -279,7 +283,7 @@ Bot::builder() For scenarios where you need to process events outside of a closure (e.g., testing, custom event loops, or runtime-agnostic code), use `ChannelEventHandler` with `register_handler` instead of `on_event`: ```rust -use wacore::types::events::{ChannelEventHandler, Event}; +use wacore::types::events::{ChannelEventHandler, Event, InboundMessage}; let mut bot = Bot::builder() .with_backend(backend) @@ -299,8 +303,10 @@ let handle = bot.run().await?; while let Ok(event) = event_rx.recv().await { match &*event { Event::Connected(_) => println!("Connected!"), - Event::Message(msg, info) => { - println!("Message from {}", info.source.sender); + Event::Messages(batch) => { + for InboundMessage { info, .. } in batch.iter() { + println!("Message from {}", info.source.sender); + } } _ => {} } @@ -672,7 +678,7 @@ If a [`with_task_instrument`](#with_task_instrument) hook is configured, `run()` ## MessageContext -A convenience helper for message handling. You can construct it from the `Event::Message` components: +A convenience helper for message handling. You can construct it from an `InboundMessage` — the item type carried by `Event::Messages`' `MessageBatch`: ```rust pub struct MessageContext { @@ -683,7 +689,7 @@ pub struct MessageContext { ``` -Since v0.6 `message` is `Arc` (was `Box`). This matches the `Event::Message` payload and lets `from_event` / `from_arc` reuse the bus-dispatched `Arc` with zero deep clones. +Since v0.6 `message` is `Arc` (was `Box`). This matches the `InboundMessage` payload and lets `from_inbound` / `from_arc` reuse the bus-dispatched `Arc` with zero deep clones. ### from_parts @@ -700,15 +706,15 @@ Constructs a `MessageContext` from individual message components. Internally clo pub fn from_arc(message: Arc, info: &MessageInfo, client: Arc) -> Self ``` -Constructs a `MessageContext` from an existing `Arc` without copying the body — pair this with the `Arc` you receive from `Event::Message` to keep dispatch zero-clone. +Constructs a `MessageContext` from an existing `Arc` without copying the body — pair this with the `Arc` you receive from an `InboundMessage` to keep dispatch zero-clone. -### from_event +### from_inbound ```rust -pub fn from_event(event: &Event, client: Arc) -> Option +pub fn from_inbound(inbound: &InboundMessage, client: Arc) -> Self ``` -Extracts a `MessageContext` from an `Event`. Returns `None` if the event is not an `Event::Message`. Reuses the existing `Arc` rather than cloning the body. +Extracts a `MessageContext` from a single `InboundMessage` (one item of a `MessageBatch`). Unlike the removed `from_event`, this is infallible — there's no "wrong event kind" case once you're iterating `Event::Messages`' batch. Reuses the existing `Arc` rather than cloning the body. This is what `Bot::on_message` uses internally to fan a batch out to your per-message handler, invoked once per item in arrival order. ### send_message @@ -737,7 +743,8 @@ use waproto::whatsapp as wa; use whatsapp_rust::bot::MessageContext; .on_event(|event, client| async move { - if let Some(ctx) = MessageContext::from_event(&event, client) { + for inbound in event.messages() { + let ctx = MessageContext::from_inbound(inbound, client.clone()); let reply = wa::Message { extended_text_message: buffa::MessageField::some(wa::message::ExtendedTextMessage { text: Some("Quoted reply!".to_string()), @@ -790,8 +797,9 @@ Internally this calls [`Client::send_reaction`](/api/send#send_reaction) with `s use whatsapp_rust::bot::MessageContext; .on_event(|event, client| async move { - if let Some(ctx) = MessageContext::from_event(&event, client) { + for inbound in event.messages() { // React with a thumbs-up to every incoming message. + let ctx = MessageContext::from_inbound(inbound, client.clone()); let _ = ctx.react("👍").await; } }) @@ -809,7 +817,7 @@ Newsletter (channel) messages don't flow through `MessageContext::react`. Use [` use whatsapp_rust::bot::Bot; use whatsapp_rust::TokioRuntime; use whatsapp_rust::store::SqliteStore; -use wacore::types::events::Event; +use wacore::types::events::{Event, InboundMessage}; use whatsapp_rust_tokio_transport::TokioWebSocketTransportFactory; use whatsapp_rust_ureq_http_client::UreqHttpClient; use waproto::whatsapp as wa; @@ -829,14 +837,16 @@ async fn main() -> anyhow::Result<()> { .skip_history_sync() // Bot only needs new messages .on_event(|event, client| async move { match &*event { - Event::Message(msg, info) => { + Event::Messages(batch) => { // Echo messages back - if let Some(text) = &msg.conversation { - let reply = wa::Message { - conversation: Some(format!("You said: {}", text)), - ..Default::default() - }; - let _ = client.send_message(info.source.chat.clone(), reply).await; + for InboundMessage { message: msg, info } in batch.iter() { + if let Some(text) = &msg.conversation { + let reply = wa::Message { + conversation: Some(format!("You said: {}", text)), + ..Default::default() + }; + let _ = client.send_message(info.source.chat.clone(), reply).await; + } } } Event::Connected(_) => { diff --git a/api/client.mdx b/api/client.mdx index 92485f96..949964c6 100644 --- a/api/client.mdx +++ b/api/client.mdx @@ -1609,7 +1609,7 @@ Registers an event handler for protocol events. **Example:** ```rust -use wacore::types::events::{Event, EventHandler}; +use wacore::types::events::{Event, EventHandler, InboundMessage}; use std::sync::Arc; struct MyHandler; @@ -1617,7 +1617,11 @@ struct MyHandler; impl EventHandler for MyHandler { fn handle_event(&self, event: Arc) { match &*event { - Event::Message(msg, info) => println!("New message from {}: {:?}", info.source.sender, msg), + Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + println!("New message from {}: {:?}", info.source.sender, msg); + } + } Event::Connected(_) => println!("Connected!"), _ => {} } diff --git a/api/polls.mdx b/api/polls.mdx index 01ddf7dc..a78bea4c 100644 --- a/api/polls.mdx +++ b/api/polls.mdx @@ -350,7 +350,7 @@ The `wacore::poll` module exposes the cryptographic primitives used internally: WhatsApp wraps message edits, poll edits, poll add-option, and event edits in a single `secret_encrypted_message` envelope keyed by a per-use-case secret derived from the parent message's `message_secret`. v0.6 added support for decrypting all four kinds. -**The client now decrypts these inline on receive.** Since v0.6 the receive path automatically resolves the parent secret from the [`MsgSecretStore`](/api/store#msgsecretstore), decrypts the envelope, and dispatches the result as a normal [`Event::Message`](/concepts/events#message) carrying the decrypted payload — so most apps never call the helpers below. The client captures `MessageContextInfo.message_secret` from inbound messages and seeds secrets from history-sync, so edits decrypt as long as the parent secret is known. Edits whose secret can't be found are skipped silently (no undecryptable event); the raw envelope stays on the message. The manual helpers remain for custom pipelines or when you store secrets yourself. +**The client now decrypts these inline on receive.** Since v0.6 the receive path automatically resolves the parent secret from the [`MsgSecretStore`](/api/store#msgsecretstore), decrypts the envelope, and dispatches the result as a normal [`Event::Messages`](/concepts/events#messages) carrying the decrypted payload — so most apps never call the helpers below. The client captures `MessageContextInfo.message_secret` from inbound messages and seeds secrets from history-sync, so edits decrypt as long as the parent secret is known. Edits whose secret can't be found are skipped silently (no undecryptable event); the raw envelope stays on the message. The manual helpers remain for custom pipelines or when you store secrets yourself. ```rust diff --git a/api/signal.mdx b/api/signal.mdx index 96d7a827..0f4605c2 100644 --- a/api/signal.mdx +++ b/api/signal.mdx @@ -305,7 +305,7 @@ When you message Meta AI or another `@bot` account, the bot's replies arrive as The client handles this end to end and **transparently**: 1. **On send to a bot**, the outbound `MessageContextInfo.messageSecret` is persisted (keyed by `(chat, sender, msg_id)`) so the reply can be decrypted later. -2. **On receive**, an `msmsg` stanza is decrypted and decoded into a `wa::Message`, then dispatched as a normal [`Event::Message`](/concepts/events#message) — there is no separate bot event. The sender is the bot JID (e.g. `…@bot`) and `MsgMetaInfo.target_id` points back at your original prompt. +2. **On receive**, an `msmsg` stanza is decrypted and decoded into a `wa::Message`, then dispatched as a normal [`Event::Messages`](/concepts/events#messages) — there is no separate bot event. The sender is the bot JID (e.g. `…@bot`) and `MsgMetaInfo.target_id` points back at your original prompt. 3. **On failure** (missing secret, GCM tag mismatch, malformed proto) the client nacks with reason `495` (`MissingMessageSecret`) instead of silently dropping, and group bot replies are acked with a bare `` matching WA Web. You don't need to call anything — receiving bot replies works as soon as you've sent a message to the bot from the same client. The low-level primitive is `wacore::bot_message::decrypt_bot_message(message_secret, enc_iv, enc_payload, ctx)`, and persistence is backed by the [`MsgSecretStore`](/api/store#msgsecretstore) trait. diff --git a/concepts/architecture.mdx b/concepts/architecture.mdx index cbec7b87..d815a32a 100644 --- a/concepts/architecture.mdx +++ b/concepts/architecture.mdx @@ -273,7 +273,8 @@ Incoming stanzas are decoded as `Arc` (zero-copy from the network pub async fn handle_message(client: &Arc, node: &Arc) { // 1. Extract encrypted message from NodeRef // 2. Decrypt via Signal Protocol - // 3. Dispatch Event::Message + // 3. Commit (or accumulate into a batch during the offline drain) + // 4. Dispatch Event::Messages } ``` @@ -421,6 +422,10 @@ pub(crate) struct OfflineSyncMetrics { **Concurrency gating:** During offline sync, the client restricts message processing to a single concurrent task (1 semaphore permit) to preserve ordering. Once sync completes — either by the server end marker, all expected items arriving, or timeout — the semaphore is expanded to 64 permits, switching to parallel message processing. + +The drain→live transition also flushes the tail of the [inbound commit batch](/advanced/inbound-durability#batching): the last accumulated batch of decrypted messages commits (buffer → Signal flush → durability hook → acks → `Event::Messages`) before the semaphore widens, so no live-mode message is processed ahead of it. If that tail commit fails, the transition is deferred and retried every 3 seconds while the client stays in single-permit drain mode; `OfflineSyncCompleted` still fires immediately so startup waiters are not blocked on the retry. + + **Semaphore transition safety:** When the semaphore is swapped from 1 to 64 permits, tasks that were already waiting on the old semaphore must not be silently dropped. The client uses a **generation-checked re-acquire loop** to handle this transition safely: 1. Each semaphore swap increments an atomic `message_semaphore_generation` counter @@ -435,7 +440,7 @@ This prevents a critical issue where `pkmsg` messages (which carry Sender Key Di **Pull-batch backlog drain (v0.6):** Offline resume now drives the same pull-batch loop WA Web uses to drain the backlog: stanzas that the client can't process (unrecognized ``, known-but-empty `` content, duplicates, ciphertexts that decrypt-fail terminally) are transport-acked alongside their retry receipt so the server stops re-delivering them. Before this, such a stanza fell through `classify_incoming_message` silently, so the server kept replaying it from the offline queue every reconnect until `` closed the stream. The drain logic also acks duplicate-message PDOs that previously hit the silent-drop branch in `handle_decrypted_plaintext`, and preserves the original `recipient` attribute (via `Client::spawn_node_transport_ack`, which echoes the raw `NodeRef` instead of rebuilding from `MessageInfo`) so LID-routed offline stanzas don't trigger `` on the ack. -The Meta AI bot's `msmsg` (``) encryption type was the original motivating case for this drain — it could not be decrypted, only acked. Since the bot-secret decryption landed (see [Bot message decryption](/api/signal#bot-message-decryption-msmsg)), `msmsg` stanzas are decrypted and dispatched as normal `Event::Message`s; the drain still covers genuinely unrecognized or undecryptable enc types. +The Meta AI bot's `msmsg` (``) encryption type was the original motivating case for this drain — it could not be decrypted, only acked. Since the bot-secret decryption landed (see [Bot message decryption](/api/signal#bot-message-decryption-msmsg)), `msmsg` stanzas are decrypted and dispatched as normal `Event::Messages`; the drain still covers genuinely unrecognized or undecryptable enc types. **Nack on terminal decrypt failure:** When a ciphertext exhausts retries (`max-retry` reached in the PDO recovery state machine), the client now emits a `` carrying a structured `NackReason` code instead of silently dropping the message. The full set of 21 codes (`ParsingError`, `InvalidProtobuf`, `MissingMessageSecret`, etc.) mirrors WA Web's reason set so the server stops retransmitting once it sees a terminal nack — see [Protocol → Nack reasons](/advanced/binary-protocol#nack-reasons). diff --git a/concepts/events.mdx b/concepts/events.mdx index aab3c5f2..4e3f2120 100644 --- a/concepts/events.mdx +++ b/concepts/events.mdx @@ -60,13 +60,18 @@ Handlers receive `Arc` — a shared reference-counted pointer to the even **Implementation:** ```rust +use wacore::types::events::{Event, InboundMessage}; +use std::sync::Arc; + struct MyHandler; impl EventHandler for MyHandler { fn handle_event(&self, event: Arc) { match &*event { - Event::Message(msg, info) => { - println!("Message from {}: {:?}", info.source.sender, msg); + Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + println!("Message from {}: {:?}", info.source.sender, msg); + } } _ => {} } @@ -87,13 +92,13 @@ impl EventHandler for MyHandler { fn handle_event(&self, event: Arc) { /* … */ } fn interest(&self) -> EventInterest { - // Only Message and Connected events reach this handler. - EventInterest::of(&[EventKind::Message, EventKind::Connected]) + // Only Messages and Connected events reach this handler. + EventInterest::of(&[EventKind::Messages, EventKind::Connected]) } } ``` -- **`EventKind`** is a `#[repr(u8)]` discriminant — one variant per `Event` variant (`Message`, `Connected`, `Receipt`, …). The enum is `#[non_exhaustive]`, so `match` blocks on `EventKind` must include a wildcard arm (`_ => …`); new kinds may be added in minor releases as the library tracks new server events. +- **`EventKind`** is a `#[repr(u8)]` discriminant — one variant per `Event` variant (`Messages`, `Connected`, `Receipt`, …). The enum is `#[non_exhaustive]`, so `match` blocks on `EventKind` must include a wildcard arm (`_ => …`); new kinds may be added in minor releases as the library tracks new server events. - **`EventKind::CAPACITY`** is a public `u8` constant (currently `64`) that bounds the number of kinds. It exists because each discriminant is packed as a bit in `EventInterest`'s `u64` mask, and a future variant that would overflow it fails compilation rather than silently corrupting the mask at runtime. Treat it as a read-only ceiling — you don't need to check it at runtime. - **`EventInterest`** is a 64-bit set of kinds. Build it with `EventInterest::of(&[…])`, `EventInterest::ALL` (the default), `EventInterest::none()`, or chain `.with(kind)`. Query it with `.wants(kind)`. - The bus exposes `has_handler_for(kind)` and only produces an event when at least one registered handler wants its kind. @@ -101,8 +106,8 @@ impl EventHandler for MyHandler { With the [`Bot`](/api/bot) builder, the same narrowing is available via `on_event_for`: ```rust -bot.on_event_for(&[EventKind::Message], |event, client| async move { - // only Message events +bot.on_event_for(&[EventKind::Messages], |event, client| async move { + // only Messages events }); ``` @@ -134,7 +139,7 @@ pub enum Event { LoggedOut(LoggedOut), // Messages - Message(Arc, Arc), + Messages(MessageBatch), Receipt(Receipt), UndecryptableMessage(UndecryptableMessage), Notification(Arc), @@ -563,16 +568,35 @@ Event::ClientOutdated(_) => { ## Message Events -### Message +### Messages -**Emitted:** For all incoming messages (text, media, etc.) +**Emitted:** For all incoming messages (text, media, etc.), one event per durable commit. ```rust -Event::Message(Arc, Arc) +Event::Messages(MessageBatch) + +pub struct InboundMessage { + pub message: Arc, + pub info: Arc, +} + +pub enum BatchOrigin { + Live, // delivered immediately, batch of one + OfflineDrain, // accumulated batch from the offline drain +} + +pub struct MessageBatch { + pub messages: Arc<[InboundMessage]>, + pub origin: BatchOrigin, +} ``` -Both the message body and `MessageInfo` are `Arc`-wrapped. The bus dispatches the same `Arc` to every handler — no deep clone on fan-out — and `Event::as_message()` returns `Option<(&Arc, &MessageInfo)>` so you can cheaply share the payload with spawned tasks or downstream channels. Before v0.6 the body was `Box`; the public guarantee changed from "owned, freely mutable" to "shared, immutable read access" — call `Arc::make_mut` (or clone the inner `wa::Message`) only if you genuinely need to mutate. +Live traffic dispatches a batch of one, so per-message latency is unchanged from the previous single-message event. During the offline drain the client accumulates decrypted messages and dispatches one `Event::Messages` per durable commit (size/byte/timeout triggers, matching WhatsApp Web's `MessageProcessorCache` — see [Inbound Durability](/advanced/inbound-durability#batching)). `MessageBatch` behaves as a collection: `batch.iter()`, `batch.len()`, `batch.is_empty()`, `batch.first()`, and `for msg in &batch` all work directly. `Event::as_messages()` returns `Option<&MessageBatch>`, and `Event::messages()` returns an iterator over the batch's `InboundMessage`s (empty for any other event kind) — use it to scan a mixed event stream without matching on `Event::Messages` first. + + + +Both the message body and `MessageInfo` are `Arc`-wrapped inside `InboundMessage`. The same `Arc` slice handed to a registered [durability hook](/advanced/inbound-durability) is what this event carries — no deep clone, and a consumer never sees a message the hook did not commit (newsletter messages and PDO placeholder recoveries are the two exceptions: they dispatch event-only, bypassing the hook). Before v0.6 the body was `Box`; the public guarantee changed from "owned, freely mutable" to "shared, immutable read access" — call `Arc::make_mut` (or clone the inner `wa::Message`) only if you genuinely need to mutate. **MessageInfo structure:** @@ -712,36 +736,38 @@ pub struct DeviceSentMeta { ```rust use waproto::whatsapp as wa; -Event::Message(msg, info) => { - println!("From: {} in {}", info.source.sender, info.source.chat); +Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + println!("From: {} in {}", info.source.sender, info.source.chat); - // Text message - if let Some(text) = &msg.conversation { - println!("Text: {}", text); - } + // Text message + if let Some(text) = &msg.conversation { + println!("Text: {}", text); + } - // Extended text (with link preview, quoted message, etc.) - if let Some(ext) = &msg.extended_text_message { - println!("Text: {}", ext.text.as_deref().unwrap_or("")); + // Extended text (with link preview, quoted message, etc.) + if let Some(ext) = &msg.extended_text_message { + println!("Text: {}", ext.text.as_deref().unwrap_or("")); - if let Some(context) = &ext.context_info { - if let Some(quoted) = &context.quoted_message { - println!("Quoted: {:?}", quoted); + if let Some(context) = &ext.context_info { + if let Some(quoted) = &context.quoted_message { + println!("Quoted: {:?}", quoted); + } } } - } - // Image message - if let Some(img) = &msg.image_message { - println!("Image: {} ({}x{})", - img.caption.as_deref().unwrap_or(""), - img.width.unwrap_or(0), - img.height.unwrap_or(0) - ); - } + // Image message + if let Some(img) = &msg.image_message { + println!("Image: {} ({}x{})", + img.caption.as_deref().unwrap_or(""), + img.width.unwrap_or(0), + img.height.unwrap_or(0) + ); + } - // Video, audio, document, sticker, etc. - // See waproto::whatsapp::Message for all types + // Video, audio, document, sticker, etc. + // See waproto::whatsapp::Message for all types + } } ``` @@ -806,7 +832,7 @@ Event::Receipt(receipt) => { - Group messages that fail with `NoSenderKeyState` (missing sender key) — dispatched before the retry receipt is sent - Messages with an `` node — view-once already viewed, hosted content, bot fanouts, or other server-side unavailability. For `ViewOnce`/`Hosted`/`Bot`, the phone never shares that content with a companion device, so the client acks the stanza directly instead of requesting it. Only the `Unknown` case goes through [PDO recovery](/guides/receiving-messages#unavailable-message-recovery-via-pdo) -When `is_unavailable` is `true`, the message had no encrypted content in the stanza. For `UnavailableType::Unknown`, the client sends a PDO request to your primary phone, and if the phone responds successfully, a follow-up `Event::Message` is dispatched with the recovered content. For `ViewOnce`, `Hosted`, and `Bot`, no PDO request is sent — that content is unrecoverable by design, so no follow-up `Event::Message` should be expected. +When `is_unavailable` is `true`, the message had no encrypted content in the stanza. For `UnavailableType::Unknown`, the client sends a PDO request to your primary phone, and if the phone responds successfully, a follow-up `Event::Messages` is dispatched with the recovered content (event-only — a PDO recovery bypasses the durability hook and the offline-drain batcher, dispatching immediately with `BatchOrigin::Live`). For `ViewOnce`, `Hosted`, and `Bot`, no PDO request is sent — that content is unrecoverable by design, so no follow-up `Event::Messages` should be expected. ```rust #[derive(Debug, Clone, Serialize)] @@ -2156,14 +2182,15 @@ Event::RawNode(node) => { ```rust use whatsapp_rust::bot::Bot; -use wacore::types::events::Event; +use wacore::types::events::{Event, InboundMessage}; let mut bot = Bot::builder() .with_backend(backend) .on_event(|event, client| async move { match &*event { - Event::Message(msg, info) => { - // Handle message + Event::Messages(batch) => { + // Handle each message in the batch + for InboundMessage { message: msg, info } in batch.iter() { /* … */ } } Event::Connected(_) => { // Handle connection @@ -2181,7 +2208,7 @@ let mut bot = Bot::builder() struct MessageHandler; impl EventHandler for MessageHandler { fn handle_event(&self, event: Arc) { - if let Event::Message(msg, info) = &*event { + for InboundMessage { message: msg, info } in event.messages() { // Handle messages } } @@ -2209,7 +2236,7 @@ client.register_handler(Arc::new(ConnectionHandler)); Events are buffered in an unbounded channel, so events fired before the receiver starts listening are not lost. ```rust -use wacore::types::events::{ChannelEventHandler, Event}; +use wacore::types::events::{ChannelEventHandler, Event, InboundMessage}; use std::sync::Arc; let (handler, event_rx) = ChannelEventHandler::new(); @@ -2222,8 +2249,10 @@ while let Ok(event) = event_rx.recv().await { println!("Connected!"); break; } - Event::Message(msg, info) => { - println!("Message from {}", info.source.sender); + Event::Messages(batch) => { + for InboundMessage { info, .. } in batch.iter() { + println!("Message from {}", info.source.sender); + } } _ => {} } @@ -2317,7 +2346,7 @@ Event::HistorySync(lazy_sync) => { ### Arc\ dispatch -With `Arc` dispatch, each event is wrapped in a single `Arc` by the `CoreEventBus` and shared across all handlers. This eliminates deep clones of large event payloads like `LazyHistorySync` blobs and `Message(Arc, Arc)`. Both the `wa::Message` body and the `MessageInfo` are `Arc`-wrapped, enabling zero-cost sharing across the message dispatch, retry receipt, and PDO recovery paths without cloning the full struct. +With `Arc` dispatch, each event is wrapped in a single `Arc` by the `CoreEventBus` and shared across all handlers. This eliminates deep clones of large event payloads like `LazyHistorySync` blobs and `Messages(MessageBatch)`. Both the `wa::Message` body and the `MessageInfo` inside each `InboundMessage` are `Arc`-wrapped, enabling zero-cost sharing across the message dispatch, durability hook, retry receipt, and PDO recovery paths without cloning the full struct — the batch handed to a registered durability hook is the very same `Arc<[InboundMessage]>` this event carries. Combined with `LazyHistorySync`'s `OnceLock`, all handlers sharing the same `Arc` get parse-once semantics for free — the first handler to call `lazy_sync.get()` triggers the decode, and subsequent handlers reuse the cached result. @@ -2328,14 +2357,15 @@ Combined with `LazyHistorySync`'s `OnceLock`, all handlers sharing the same `Arc ```rust .on_event(|event, client| async move { // Only handle events you care about - match &*event { - Event::Message(msg, info) if info.source.is_group => { + for InboundMessage { message: msg, info } in event.messages() { + // These are independent filters, not mutually exclusive cases — + // a non-self-sent group message enters both blocks. + if info.source.is_group { // Only handle group messages } - Event::Message(msg, info) if !info.source.is_from_me => { + if !info.source.is_from_me { // Only handle messages from others } - _ => {} } }) ``` @@ -2350,11 +2380,8 @@ Combined with `LazyHistorySync`'s `OnceLock`, all handlers sharing the same `Arc }) async fn handle_event(event: &Event, client: Arc) -> Result<()> { - match event { - Event::Message(msg, info) => { - process_message(msg, info, client).await? - } - _ => {} + for InboundMessage { message: msg, info } in event.messages() { + process_message(msg, info, client.clone()).await? } Ok(()) } @@ -2364,12 +2391,12 @@ async fn handle_event(event: &Event, client: Arc) -> Result<()> { ```rust .on_event(|event, client| async move { - if let Event::Message(msg, info) = &*event { + if matches!(&*event, Event::Messages(_)) { let client = client.clone(); let event = event.clone(); // Arc clone — O(1) tokio::spawn(async move { - if let Event::Message(msg, info) = &*event { + for InboundMessage { message: msg, info } in event.messages() { process_message(msg, info, &client).await; } }); diff --git a/guides/communities.mdx b/guides/communities.mdx index a67dbb89..14701a5a 100644 --- a/guides/communities.mdx +++ b/guides/communities.mdx @@ -251,10 +251,12 @@ For CAG chats the library checks `GroupInfo::is_community_announce` (populated f Incoming encrypted reactions from CAG posts are decrypted transparently by the receive path and surfaced as a normal `reaction_message` event. The `key` field is filled from the envelope's `target_message_key`, so your event handler looks identical to a regular group reaction: ```rust -Event::Message(msg, info) => { - if let Some(reaction) = msg.reaction_message.as_option() { - println!("Reaction: {:?}", reaction.text); - println!("On post: {:?}", reaction.key.as_option().and_then(|k| k.id.as_deref())); +Event::Messages(batch) => { + for InboundMessage { message: msg, .. } in batch.iter() { + if let Some(reaction) = msg.reaction_message.as_option() { + println!("Reaction: {:?}", reaction.text); + println!("On post: {:?}", reaction.key.as_option().and_then(|k| k.id.as_deref())); + } } } ``` @@ -305,16 +307,18 @@ Each comment carries a fresh `messageSecret` of its own so it can receive encryp ### Receiving comments -Incoming encrypted comments are decrypted transparently on the receive path. The decrypted body is dispatched as a normal `Event::Message`. Because the inner `Message` proto has no slot for the parent post key, the threading link surfaces on `MessageInfo::comment_target`: +Incoming encrypted comments are decrypted transparently on the receive path. The decrypted body is dispatched as part of a normal `Event::Messages` batch. Because the inner `Message` proto has no slot for the parent post key, the threading link surfaces on `MessageInfo::comment_target`: ```rust -Event::Message(msg, info) => { - if let Some(parent_key) = &info.comment_target { - // This is a channel comment. - println!("Comment on post: {:?}", parent_key.id); - - if let Some(text) = msg.text_content() { - println!("Comment text: {}", text); +Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + if let Some(parent_key) = &info.comment_target { + // This is a channel comment. + println!("Comment on post: {:?}", parent_key.id); + + if let Some(text) = msg.text_content() { + println!("Comment text: {}", text); + } } } } diff --git a/guides/custom-backends.mdx b/guides/custom-backends.mdx index 822cba46..f5aeb071 100644 --- a/guides/custom-backends.mdx +++ b/guides/custom-backends.mdx @@ -468,8 +468,51 @@ See [Store API reference](/api/store#protocol-store) for all ProtocolStore metho If you want to support the opt-in [`InboundDurabilityHook`](/advanced/inbound-durability), your backend must implement four additional `ProtocolStore` methods. Without them, `Bot::build()` rejects any attempt to register a durability hook with a `BotBuilderError::UnsupportedDurabilityBackend` error. + +Two further methods, `store_pending_inbound_batch` and `delete_pending_inbound_batch`, back the client's [inbound commit batching](/advanced/inbound-durability#batching) during the offline drain. Both default to looping the single-row methods below, so implementing just the four methods in this section is enough — your backend keeps working, just without batch atomicity. Override them if you want the whole batch committed in one transaction: + +```rust +async fn store_pending_inbound_batch(&self, rows: &[PendingInboundRow<'_>]) -> Result<()> { + // device_id is *your* store's own field (like SqliteStore's), not a + // parameter — see the note on device_id scoping below. + let device_id = self.device_id; + // e.g. one multi-row INSERT inside a single transaction + self.connection.transaction(|tx| async move { + for row in rows { + tx.execute( + "INSERT OR REPLACE INTO pending_inbound_messages \ + (chat, sender, id, device_id, message, inserted_at) \ + VALUES (?, ?, ?, ?, ?, strftime('%s', 'now'))", + &[row.chat, row.sender, row.id, &device_id, row.message], + ).await?; + } + Ok(()) + }).await +} + +async fn delete_pending_inbound_batch(&self, keys: &[PendingInboundKey<'_>]) -> Result<()> { + let device_id = self.device_id; + self.connection.transaction(|tx| async move { + for key in keys { + tx.execute( + "DELETE FROM pending_inbound_messages WHERE chat = ? AND sender = ? AND id = ? AND device_id = ?", + &[key.chat, key.sender, key.id, &device_id], + ).await?; + } + Ok(()) + }).await +} +``` + +The bundled `SqliteStore` overrides both this way, so the offline-drain batcher pays one Diesel transaction per batch instead of one round-trip per message. + + The default trait implementations **fail closed** (return an error) rather than silently degrading to at-most-once, so you only need to override them if your backend opts in. + +`store_pending_inbound`/`get_pending_inbound`/`delete_pending_inbound`/`delete_expired_pending_inbound` and their batch counterparts do **not** take `device_id` as a parameter — `(chat, sender, id)` is the whole key on the trait. A store that serves a single account scopes rows by `(chat, sender, id)` alone. If your backend is a **shared multi-account store** (one table/connection serving several `SqliteStore`-like instances), scope rows by your own instance's device id internally, the same way the bundled `SqliteStore` reads its own `self.device_id` field rather than accepting it as an argument. + + ```rust #[async_trait] impl ProtocolStore for MyCustomStore { @@ -477,18 +520,19 @@ impl ProtocolStore for MyCustomStore { // --- Pending Inbound Buffer (required for InboundDurabilityHook) --- // - // Keyed by (chat, sender, id, device_id): stanza ids are only unique - // within a (chat, sender) pair; device_id isolates rows across accounts - // in shared multi-device stores. + // Keyed by (chat, sender, id): stanza ids are only unique within a + // (chat, sender) pair. `self.device_id` scopes rows to this store + // instance for backends sharing one table across accounts/devices — + // it is not part of the trait signature. async fn store_pending_inbound( &self, chat: &str, sender: &str, id: &str, - device_id: u32, message: &[u8], ) -> Result<()> { + let device_id = self.device_id; self.connection.execute( "INSERT OR REPLACE INTO pending_inbound_messages \ (chat, sender, id, device_id, message, inserted_at) \ @@ -503,8 +547,8 @@ impl ProtocolStore for MyCustomStore { chat: &str, sender: &str, id: &str, - device_id: u32, ) -> Result>> { + let device_id = self.device_id; let row = self.connection.query_optional( "SELECT message FROM pending_inbound_messages \ WHERE chat = ? AND sender = ? AND id = ? AND device_id = ?", @@ -518,8 +562,8 @@ impl ProtocolStore for MyCustomStore { chat: &str, sender: &str, id: &str, - device_id: u32, ) -> Result<()> { + let device_id = self.device_id; self.connection.execute( "DELETE FROM pending_inbound_messages \ WHERE chat = ? AND sender = ? AND id = ? AND device_id = ?", @@ -530,11 +574,8 @@ impl ProtocolStore for MyCustomStore { // Called unconditionally by the keepalive sweep on every backend. // Must return Ok(0) for backends that have no buffer rather than erroring. - async fn delete_expired_pending_inbound( - &self, - device_id: u32, - cutoff_timestamp: i64, - ) -> Result { + async fn delete_expired_pending_inbound(&self, cutoff_timestamp: i64) -> Result { + let device_id = self.device_id; let count = self.connection.execute( "DELETE FROM pending_inbound_messages \ WHERE device_id = ? AND inserted_at < ?", @@ -563,6 +604,10 @@ CREATE INDEX idx_pending_inbound_inserted ON pending_inbound_messages (device_id, inserted_at); ``` + +The `device_id` column exists so one shared table can serve multiple accounts/devices without cross-account collisions — the bundled `SqliteStore` has one such table per account and always filters by its own `self.device_id`. A backend with one table per account can drop the column and the `device_id` predicate entirely. + + The `message` column stores the proto-serialized `wa::Message` bytes. The client writes them before the Signal ratchet advances so that a crash between the buffer write and the hook return can replay the exact same decrypted bytes on redelivery. diff --git a/guides/media-handling.mdx b/guides/media-handling.mdx index b9391126..2e68c38c 100644 --- a/guides/media-handling.mdx +++ b/guides/media-handling.mdx @@ -16,39 +16,42 @@ Download media directly from message types that implement `Downloadable`: ```rust use wacore::download::Downloadable; use whatsapp_rust::client::Client; +use wacore::types::events::InboundMessage; match event { - Event::Message(message, info) => { - // Image - if let Some(img) = message.image_message.as_option() { - let data = client.download(img).await?; - std::fs::write("image.jpg", data)?; - } - - // Video - if let Some(video) = message.video_message.as_option() { - let data = client.download(video).await?; - std::fs::write("video.mp4", data)?; - } - - // Audio - if let Some(audio) = message.audio_message.as_option() { - let data = client.download(audio).await?; - let ext = if audio.ptt() { "ogg" } else { "mp3" }; - std::fs::write(format!("audio.{}", ext), data)?; - } - - // Document - if let Some(doc) = message.document_message.as_option() { - let data = client.download(doc).await?; - let filename = doc.file_name.as_deref().unwrap_or("document"); - std::fs::write(filename, data)?; - } - - // Sticker - if let Some(sticker) = message.sticker_message.as_option() { - let data = client.download(sticker).await?; - std::fs::write("sticker.webp", data)?; + Event::Messages(batch) => { + for InboundMessage { message, info } in batch.iter() { + // Image + if let Some(img) = message.image_message.as_option() { + let data = client.download(img).await?; + std::fs::write("image.jpg", data)?; + } + + // Video + if let Some(video) = message.video_message.as_option() { + let data = client.download(video).await?; + std::fs::write("video.mp4", data)?; + } + + // Audio + if let Some(audio) = message.audio_message.as_option() { + let data = client.download(audio).await?; + let ext = if audio.ptt() { "ogg" } else { "mp3" }; + std::fs::write(format!("audio.{}", ext), data)?; + } + + // Document + if let Some(doc) = message.document_message.as_option() { + let data = client.download(doc).await?; + let filename = doc.file_name.as_deref().unwrap_or("document"); + std::fs::write(filename, data)?; + } + + // Sticker + if let Some(sticker) = message.sticker_message.as_option() { + let data = client.download(sticker).await?; + std::fs::write("sticker.webp", data)?; + } } } _ => {} diff --git a/guides/receiving-messages.mdx b/guides/receiving-messages.mdx index 6dd54a5c..c54f5610 100644 --- a/guides/receiving-messages.mdx +++ b/guides/receiving-messages.mdx @@ -62,8 +62,9 @@ See [Bot API reference](/api/bot#event-handling) for full details. ```rust pub enum Event { - /// Successfully decrypted message - Message(Arc, Arc), + /// One or more successfully decrypted messages (batch of one on live + /// traffic, an accumulated batch during the offline drain) + Messages(MessageBatch), /// Message that couldn't be decrypted UndecryptableMessage(UndecryptableMessage), @@ -185,19 +186,21 @@ See [WAProto API reference](/api/waproto) for the full message type hierarchy. ```rust match event { - Event::Message(message, info) => { - // Simple text - if let Some(text) = &message.conversation { - println!("Text: {}", text); - } - - // Extended text (with links, formatting) - if let Some(ext) = &message.extended_text_message { - if let Some(text) = &ext.text { - println!("Extended text: {}", text); + Event::Messages(batch) => { + for InboundMessage { message, info } in batch.iter() { + // Simple text + if let Some(text) = &message.conversation { + println!("Text: {}", text); } - if let Some(url) = &ext.matched_text { - println!("Contains link: {}", url); + + // Extended text (with links, formatting) + if let Some(ext) = &message.extended_text_message { + if let Some(text) = &ext.text { + println!("Extended text: {}", text); + } + if let Some(url) = &ext.matched_text { + println!("Contains link: {}", url); + } } } } @@ -263,17 +266,19 @@ if let Some(reaction) = &message.reaction_message { ### Channel Comments -Encrypted channel comments from Community Announcement Groups are decrypted transparently and dispatched as `Event::Message` carrying the comment body. The parent post key surfaces on `MessageInfo::comment_target` (the inner `Message` proto has no slot for the threading link): +Encrypted channel comments from Community Announcement Groups are decrypted transparently and dispatched as `Event::Messages` carrying the comment body. The parent post key surfaces on `MessageInfo::comment_target` (the inner `Message` proto has no slot for the threading link): ```rust -Event::Message(msg, info) => { - if let Some(parent_key) = &info.comment_target { - // This Event::Message is a decrypted CAG channel comment. - println!("Comment on post: {:?}", parent_key.id); - println!("Post author: {:?}", parent_key.participant); - - if let Some(text) = msg.text_content() { - println!("Comment text: {}", text); +Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + if let Some(parent_key) = &info.comment_target { + // This item is a decrypted CAG channel comment. + println!("Comment on post: {:?}", parent_key.id); + println!("Post author: {:?}", parent_key.participant); + + if let Some(text) = msg.text_content() { + println!("Comment text: {}", text); + } } } } @@ -309,18 +314,20 @@ When you send a message from one device, other devices receive it as a `DeviceSe ```rust // The library handles this automatically - you receive the inner message directly match event { - Event::Message(message, info) => { - // If this was originally a DeviceSentMessage, the library has: - // 1. Extracted the inner message content - // 2. Merged message_context_info from outer + inner - // - message_secret: inner value, fallback to outer - // - limit_sharing_v2: always from outer - // - thread_id: inner if non-empty, otherwise outer - // - bot_metadata: inner value, fallback to outer - - // You can safely access the merged context - if let Some(ctx) = &message.message_context_info { - println!("Thread: {:?}", ctx.thread_id); + Event::Messages(batch) => { + for InboundMessage { message, info } in batch.iter() { + // If this was originally a DeviceSentMessage, the library has: + // 1. Extracted the inner message content + // 2. Merged message_context_info from outer + inner + // - message_secret: inner value, fallback to outer + // - limit_sharing_v2: always from outer + // - thread_id: inner if non-empty, otherwise outer + // - bot_metadata: inner value, fallback to outer + + // You can safely access the merged context + if let Some(ctx) = &message.message_context_info { + println!("Thread: {:?}", ctx.thread_id); + } } } _ => {} @@ -338,11 +345,13 @@ Self-sent messages synced from your primary device are automatically unwrapped. Messages are automatically decrypted by the client: ```rust -// The Event::Message already contains decrypted content +// The Event::Messages batch already contains decrypted content match event { - Event::Message(message, info) => { - // Message is already decrypted and ready to use - println!("Decrypted: {:?}", message.conversation); + Event::Messages(batch) => { + for InboundMessage { message, .. } in batch.iter() { + // Message is already decrypted and ready to use + println!("Decrypted: {:?}", message.conversation); + } } _ => {} } @@ -449,9 +458,9 @@ Only a plain (`Unknown`) fanout is recovered via PDO. The flow for that case is: 2. An `UndecryptableMessage` event is dispatched immediately with `is_unavailable: true` 3. A PDO request (`PlaceholderMessageResend`) is sent to your own bare JID (server routes to all devices including device 0) 4. The phone responds with the full `WebMessageInfo` containing the decrypted message -5. The client validates the response came from device 0 (primary phone) and dispatches the recovered message as a normal `Event::Message` +5. The client validates the response came from device 0 (primary phone) and dispatches the recovered message as a normal `Event::Messages` — event-only, bypassing the durability hook and the offline-drain batcher (delivered immediately, `BatchOrigin::Live`) -For `ViewOnce`, `Hosted`, and `Bot`, only steps 1–2 happen: the client dispatches `UndecryptableMessage` and acks immediately. There is no PDO round-trip and no follow-up `Event::Message` to wait for. +For `ViewOnce`, `Hosted`, and `Bot`, only steps 1–2 happen: the client dispatches `UndecryptableMessage` and acks immediately. There is no PDO round-trip and no follow-up `Event::Messages` to wait for. The recovered `MessageInfo` (for the PDO-recovered `Unknown` case) includes `unavailable_request_id` — the PDO request message ID — so you can correlate recovered messages with the original `UndecryptableMessage` event. @@ -461,7 +470,7 @@ Event::UndecryptableMessage(undec) => { match undec.unavailable_type { UnavailableType::Unknown => { // Content is being requested from your phone via PDO. - // You'll receive an Event::Message when the phone responds. + // You'll receive an Event::Messages when the phone responds. println!("Unavailable message — requesting from phone"); } UnavailableType::ViewOnce => { @@ -479,9 +488,11 @@ Event::UndecryptableMessage(undec) => { // When the phone responds to a PDO request (Unknown fanouts only), the // recovered message includes the request ID: -Event::Message(msg, info) => { - if let Some(request_id) = &info.unavailable_request_id { - println!("Recovered via PDO (request: {})", request_id); +Event::Messages(batch) => { + for InboundMessage { info, .. } in batch.iter() { + if let Some(request_id) = &info.unavailable_request_id { + println!("Recovered via PDO (request: {})", request_id); + } } } ``` @@ -565,7 +576,7 @@ match event { By default, the client acknowledges a message to the server as soon as it is decrypted. If your process crashes before you persist the message, it is lost — the server will not redeliver it. -Register an `InboundDurabilityHook` to defer the ack until your consumer durably commits the message: +Register an `InboundDurabilityHook` to defer the ack until your consumer durably commits the message(s). Live traffic calls the hook with a batch of one; an offline drain hands over an accumulated batch (WhatsApp Web's `MessageProcessorCache` granularity), so the durability cost amortizes over the batch instead of paying a round-trip per message: ```rust use std::sync::Arc; @@ -577,15 +588,19 @@ struct MyStore; #[async_trait] impl InboundDurabilityHook for MyStore { - async fn on_message( + async fn on_messages( &self, _client: Arc, - info: &MessageInfo, - message: &wa::Message, + batch: &[InboundMessage], ) -> anyhow::Result<()> { - // Return Ok only after the commit is durable. - // Return Err to suppress the ack and trigger redelivery. - my_db_insert(&info.id, message).await?; + // This loop commits each item independently, so a later item's `?` + // can leave earlier items already committed while the SDK still + // redelivers the whole batch on Err. `my_db_insert` MUST be an + // idempotent upsert (e.g. `INSERT ... ON CONFLICT DO NOTHING`), or + // wrap the loop in one transaction — see the note below. + for item in batch { + my_db_insert(&item.info.id, &item.message).await?; + } Ok(()) } } @@ -597,9 +612,9 @@ let bot = Bot::builder() .await?; ``` -The hook **must be idempotent** — deduplicate by `(info.source.chat, info.source.sender, info.id)` since a crash after the consumer commits but before the ack lands will replay the message. +The hook **must be idempotent** — deduplicate by `(info.source.chat, info.source.sender, info.id)` since a crash after the consumer commits but before the ack lands will replay the message, and a failed batch is redelivered whole. -See [Inbound Durability Hook](/advanced/inbound-durability) for the full contract, caveats, and a worked example. +See [Inbound Durability Hook](/advanced/inbound-durability) for the full contract, batching triggers, caveats, and a worked example. ### Custom encryption handlers @@ -695,10 +710,12 @@ See [Signal Protocol](/advanced/signal-protocol) for more on session management. ```rust .on_event(|event, client| async move { match &*event { - Event::Message(message, info) => { - // Process message - if let Err(e) = process_message(message, info, client).await { - eprintln!("Error processing message {}: {:?}", info.id, e); + Event::Messages(batch) => { + // Process each message in the batch + for InboundMessage { message, info } in batch.iter() { + if let Err(e) = process_message(message, info, client.clone()).await { + eprintln!("Error processing message {}: {:?}", info.id, e); + } } } Event::UndecryptableMessage(undecryptable) => { @@ -754,7 +771,7 @@ Always handle critical events: ```rust match &*event { - Event::Message(message, info) => { /* ... */ } + Event::Messages(batch) => { /* ... */ } Event::Connected(_) => { /* Initialize */ } Event::Disconnected(_) => { /* Cleanup */ } Event::LoggedOut(_) => { /* Re-authenticate */ } @@ -768,13 +785,13 @@ Spawn tasks for long-running operations: ```rust .on_event(|event, client| async move { - if let Event::Message(message, info) = &*event { + if matches!(&*event, Event::Messages(_)) { // Spawn task for heavy processing — Arc clone is O(1) let client = client.clone(); let event = event.clone(); tokio::spawn(async move { - if let Event::Message(message, info) = &*event { - process_heavy_task(message, info, client).await; + for InboundMessage { message, info } in event.messages() { + process_heavy_task(message, info, client.clone()).await; } }); } diff --git a/guides/sending-messages.mdx b/guides/sending-messages.mdx index ccc5b06c..fe83fed2 100644 --- a/guides/sending-messages.mdx +++ b/guides/sending-messages.mdx @@ -234,7 +234,8 @@ If you're already inside an event handler, [`MessageContext::react`](/api/bot#re ```rust use whatsapp_rust::bot::MessageContext; -if let Some(ctx) = MessageContext::from_event(&event, client) { +for inbound in event.messages() { + let ctx = MessageContext::from_inbound(inbound, client.clone()); ctx.react("❤️").await?; } ``` @@ -304,14 +305,16 @@ let result = client.comments() The `parent_key.participant` field must identify the post author so receivers can derive the HKDF decryption key from the envelope. When `from_me` is `true` and `participant` is absent the library resolves the author to your own identity. -Incoming encrypted comments are decrypted transparently. The comment body is dispatched as `Event::Message` and the parent post key is available on `MessageInfo::comment_target`: +Incoming encrypted comments are decrypted transparently. The comment body is dispatched as part of an `Event::Messages` batch and the parent post key is available on `MessageInfo::comment_target`: ```rust -Event::Message(msg, info) => { - if let Some(parent_key) = &info.comment_target { - println!("Comment on post: {:?}", parent_key.id); - if let Some(text) = msg.text_content() { - println!("Text: {}", text); +Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + if let Some(parent_key) = &info.comment_target { + println!("Comment on post: {:?}", parent_key.id); + if let Some(text) = msg.text_content() { + println!("Text: {}", text); + } } } } @@ -693,9 +696,11 @@ let expiration = metadata.ephemeral_expiration; // 0 if disabled For incoming messages, read it from `MessageInfo`: ```rust -Event::Message(message, info) => { - if let Some(expiration) = info.ephemeral_expiration { - println!("Chat has {}s disappearing timer", expiration); +Event::Messages(batch) => { + for InboundMessage { info, .. } in batch.iter() { + if let Some(expiration) = info.ephemeral_expiration { + println!("Chat has {}s disappearing timer", expiration); + } } } ``` diff --git a/pt/quickstart.mdx b/pt/quickstart.mdx index 799a8c6c..291f3b39 100644 --- a/pt/quickstart.mdx +++ b/pt/quickstart.mdx @@ -16,7 +16,7 @@ use whatsapp_rust::TokioRuntime; use whatsapp_rust::store::SqliteStore; use whatsapp_rust_tokio_transport::TokioWebSocketTransportFactory; use whatsapp_rust_ureq_http_client::UreqHttpClient; -use wacore::types::events::Event; +use wacore::types::events::{Event, InboundMessage}; #[tokio::main] async fn main() -> Result<(), Box> { @@ -34,8 +34,10 @@ async fn main() -> Result<(), Box> { Event::PairingQrCode { code, .. } => { println!("Scan this QR code with WhatsApp:\n{}", code); } - Event::Message(msg, info) => { - println!("Message from {}: {:?}", info.source.sender, msg); + Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + println!("Message from {}: {:?}", info.source.sender, msg); + } } _ => {} } @@ -88,8 +90,10 @@ async fn main() -> Result<(), Box> { Event::PairingQrCode { code, .. } => { println!("QR Code:\n{}", code); } - Event::Message(msg, info) => { - // Trate a mensagem recebida + Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + // Trate a mensagem recebida + } } Event::Connected(_) => { println!("Connected successfully!"); @@ -134,19 +138,21 @@ use waproto::whatsapp as wa; Event::PairingQrCode { code, .. } => { println!("QR Code:\n{}", code); } - Event::Message(msg, info) => { - // Verifica se a mensagem é um texto dizendo "ping" - if let Some(text) = msg.text_content() { - if text == "ping" { - // Cria a mensagem de resposta - let reply = wa::Message { - conversation: Some("pong".to_string()), - ..Default::default() - }; - - // Envia a resposta - if let Err(e) = client.send_message(info.source.chat.clone(), reply).await { - eprintln!("Failed to send reply: {}", e); + Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + // Verifica se a mensagem é um texto dizendo "ping" + if let Some(text) = msg.text_content() { + if text == "ping" { + // Cria a mensagem de resposta + let reply = wa::Message { + conversation: Some("pong".to_string()), + ..Default::default() + }; + + // Envia a resposta + if let Err(e) = client.send_message(info.source.chat.clone(), reply).await { + eprintln!("Failed to send reply: {}", e); + } } } } @@ -285,7 +291,8 @@ Para um tratamento de mensagens mais limpo, use `MessageContext` para encapsular use whatsapp_rust::bot::{Bot, MessageContext}; .on_event(|event, client| async move { - if let Some(ctx) = MessageContext::from_event(&event, client) { + for inbound in event.messages() { + let ctx = MessageContext::from_inbound(inbound, client.clone()); handle_message(&ctx).await; } }) @@ -346,7 +353,7 @@ use chrono::{Local, Utc}; use log::{error, info}; use std::sync::Arc; use wacore::proto_helpers::MessageExt; -use wacore::types::events::Event; +use wacore::types::events::{Event, InboundMessage}; use waproto::whatsapp as wa; use whatsapp_rust::bot::{Bot, MessageContext}; use whatsapp_rust::TokioRuntime; @@ -386,9 +393,11 @@ async fn main() -> Result<(), Box> { Event::PairingQrCode { code, .. } => { println!("\n{}", code); } - Event::Message(msg, info) => { - let ctx = MessageContext::from_parts(msg, info, client); - handle_message(&ctx).await; + Event::Messages(batch) => { + for InboundMessage { message: msg, info } in batch.iter() { + let ctx = MessageContext::from_parts(msg, info, client.clone()); + handle_message(&ctx).await; + } } Event::Connected(_) => info!("Bot connected!"), Event::LoggedOut(_) => error!("Bot was logged out!"),