feat: add awaited pre-ACK message hook

src/bot.rs

@@ -487,6 +487,7 @@ pub struct BotBuilder<
     // Optional fields
     event_handlers: Vec<RegisteredHandler>,
     raw_handlers: Vec<Arc<dyn EventHandler>>,
+    pre_ack_message_hook: Option<crate::message::ParsedMessagePreAckHook>,
     custom_enc_handlers: HashMap<String, Arc<dyn EncHandler>>,
     override_version: Option<(u32, u32, u32)>,
     device_props_override: Option<DevicePropsOverride>,
@@ -508,6 +509,7 @@ impl BotBuilder<MissingBackend, DefaultTransportState, DefaultHttpState, Default
             runtime: default_runtime(),
             event_handlers: Vec::new(),
             raw_handlers: Vec::new(),
+            pre_ack_message_hook: None,
             custom_enc_handlers: HashMap::new(),
             override_version: None,
             device_props_override: None,
@@ -533,6 +535,7 @@ impl<B, T, H, R> BotBuilder<B, T, H, R> {
             runtime: self.runtime,
             event_handlers: self.event_handlers,
             raw_handlers: self.raw_handlers,
+            pre_ack_message_hook: self.pre_ack_message_hook,
             custom_enc_handlers: self.custom_enc_handlers,
             override_version: self.override_version,
             device_props_override: self.device_props_override,
@@ -639,6 +642,27 @@ impl<B, T, H, R> BotBuilder<B, T, H, R> {
         })
     }
 
+    /// Run `hook` inline before acknowledging each parsed inbound message.
+    ///
+    /// This is the durability hook for consumers that need strict
+    /// "ACK-after-commit" semantics. The hook receives the same normalized
+    /// parsed message and [`MessageInfo`] that will be emitted as
+    /// `Event::Message` after ACK. Return `Ok(())` only after your durable write
+    /// has committed. Returning `Err` suppresses both the ACK/receipt and the
+    /// normal message event for this delivery attempt so WhatsApp can retry.
+    ///
+    /// Keep this hook narrow and fast: it runs on the receive lane before ACK.
+    pub fn on_pre_ack_message<F, Fut>(mut self, hook: F) -> Self
+    where
+        F: Fn(crate::message::ParsedMessagePreAckContext) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = anyhow::Result<()>> + Send + 'static,
+    {
+        self.pre_ack_message_hook = Some(Arc::new(
+            move |ctx| -> crate::message::ParsedMessagePreAckHookFuture { Box::pin(hook(ctx)) },
+        ));
+        self
+    }
+
     /// Run `handler` with the QR payload (and validity window) each time a
     /// pairing QR code is issued. Render `code` as a QR image for scanning.
     pub fn on_qr_code<F, Fut>(self, handler: F) -> Self
@@ -969,6 +993,10 @@ impl BotBuilder<Provided, Provided, Provided, Provided> {
         // map once; the receive hot path then reads it lock-free.
         let _ = client.custom_enc_handlers.set(self.custom_enc_handlers);
 
+        if let Some(hook) = self.pre_ack_message_hook {
+            let _ = client.set_parsed_message_pre_ack_hook_arc(hook);
+        }
+
         if self.skip_history_sync {
             client.set_skip_history_sync(true);
         }

src/client.rs

@@ -501,6 +501,16 @@ pub struct Client {
     /// in `WAWebMessageProcessPlaceholder`.
     pub(crate) undecryptable_dispatched: Cache<ChatMessageId, ()>,
 
+    /// Parsed messages whose pre-ACK hook failed after Signal state advanced.
+    /// A same-process server redelivery may then arrive as a Signal duplicate;
+    /// this cache lets that duplicate retry the durable hook with the original
+    /// normalized message instead of ACKing an uncommitted delivery.
+    pub(crate) pending_parsed_message_pre_ack:
+        Cache<ChatMessageId, crate::message::PendingParsedMessagePreAck>,
+    /// Reliable barrier for Signal flush deferral. Kept separate from the cache
+    /// because `PortableCache::entry_count()` is intentionally best-effort.
+    pub(crate) pending_parsed_message_pre_ack_count: AtomicUsize,
+
     pub enable_auto_reconnect: Arc<AtomicBool>,
     pub auto_reconnect_errors: Arc<AtomicU32>,
 
@@ -563,6 +573,11 @@ pub struct Client {
     /// `OnceLock::get` (no lock) and no per-node guard acquisition.
     pub custom_enc_handlers: std::sync::OnceLock<HashMap<String, Arc<dyn EncHandler>>>,
 
+    /// Optional awaited hook for consumers that need to durably commit a parsed
+    /// inbound message before the SDK sends the delivery receipt / transport ACK.
+    pub(crate) parsed_message_pre_ack_hook:
+        std::sync::OnceLock<crate::message::ParsedMessagePreAckHook>,
+
     /// Chat state (typing indicator) handlers registered by external consumers.
     /// Each handler receives a `ChatStateEvent` describing the chat, optional participant and state.
     pub(crate) chatstate_handlers: Arc<RwLock<Vec<ChatStateHandler>>>,

src/client/accessors.rs

@@ -23,6 +23,50 @@ impl Client {
         self.core.event_bus.add_handler(handler);
     }
 
+    /// Register an awaited hook that runs after an inbound message has been
+    /// parsed and normalized, but before the SDK sends its delivery receipt or
+    /// transport ACK.
+    ///
+    /// The hook is intended for durability-sensitive consumers: persist the
+    /// message, commit the transaction, then return `Ok(())`. If the hook
+    /// returns an error, `whatsapp-rust` suppresses the ACK and the normal
+    /// `Event::Message` dispatch for this attempt, leaving the server free to
+    /// redeliver/retry instead of marking the message delivered.
+    ///
+    /// Only one hook can be registered for a `Client`. Applications using
+    /// [`BotBuilder`](crate::bot::BotBuilder) can prefer
+    /// [`BotBuilder::on_pre_ack_message`](crate::bot::BotBuilder::on_pre_ack_message)
+    /// to wire the hook during construction.
+    ///
+    /// Registering this hook also keeps inbound message processing serial. That
+    /// preserves the durability contract when a hook fails after Signal decrypt:
+    /// the SDK can defer Signal cache flushes while the uncommitted parsed
+    /// message is pending, without allowing another inbound message to race the
+    /// pending retry path.
+    pub fn set_parsed_message_pre_ack_hook<F, Fut>(
+        &self,
+        hook: F,
+    ) -> Result<(), crate::message::ParsedMessagePreAckHookAlreadySet>
+    where
+        F: Fn(crate::message::ParsedMessagePreAckContext) -> Fut + Send + Sync + 'static,
+        Fut: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
+    {
+        self.set_parsed_message_pre_ack_hook_arc(Arc::new(
+            move |ctx| -> crate::message::ParsedMessagePreAckHookFuture { Box::pin(hook(ctx)) },
+        ))
+    }
+
+    pub(crate) fn set_parsed_message_pre_ack_hook_arc(
+        &self,
+        hook: crate::message::ParsedMessagePreAckHook,
+    ) -> Result<(), crate::message::ParsedMessagePreAckHookAlreadySet> {
+        self.parsed_message_pre_ack_hook
+            .set(hook)
+            .map_err(|_| crate::message::ParsedMessagePreAckHookAlreadySet)?;
+        self.swap_message_semaphore(1);
+        Ok(())
+    }
+
     /// Enable or disable raw node forwarding.
     /// When enabled, `Event::RawNode` is emitted for every decoded stanza before
     /// the stanza router dispatches it. Only enable when external consumers need

src/client/adapters.rs

@@ -59,6 +59,17 @@ impl Client {
     /// Flush the in-memory signal cache to the database backend.
     /// Called after each message is decrypted or after encryption operations.
     pub(crate) async fn flush_signal_cache(&self) -> Result<(), anyhow::Error> {
+        if self.parsed_message_pre_ack_hook.get().is_some()
+            && self
+                .pending_parsed_message_pre_ack_count
+                .load(Ordering::Acquire)
+                > 0
+        {
+            return Err(anyhow::anyhow!(
+                "Signal cache flush deferred while parsed-message pre-ACK commit is pending"
+            ));
+        }
+
         // Hold no device guard across the flush: this per-message batched SQLite
         // write would otherwise block every concurrent Device write for its duration.
         let backend = self

src/client/lifecycle.rs

@@ -192,6 +192,13 @@ impl Client {
             ),
 
             undecryptable_dispatched: cache_config.undecryptable_dispatched.build_with_ttl(),
+            // Intentionally unbounded/no-TTL: an entry means Signal decrypt may
+            // have advanced volatile state for a message the application has
+            // not durably committed. Expiring it would allow a later Signal
+            // flush to make that advance durable and turn server redelivery
+            // into an ACKed duplicate without rerunning the hook.
+            pending_parsed_message_pre_ack: Cache::builder().build(),
+            pending_parsed_message_pre_ack_count: AtomicUsize::new(0),
 
             offline_sync_metrics: Arc::new(OfflineSyncMetrics {
                 active: AtomicBool::new(false),
@@ -226,6 +233,7 @@ impl Client {
             pairing_cancellation_tx: Arc::new(Mutex::new(None)),
             pair_code_state: Arc::new(Mutex::new(wacore::pair_code::PairCodeState::default())),
             custom_enc_handlers: std::sync::OnceLock::new(),
+            parsed_message_pre_ack_hook: std::sync::OnceLock::new(),
             chatstate_handlers: Arc::new(RwLock::new(Vec::new())),
             pdo_pending_requests: cache_config.pdo_pending_requests.build_with_ttl(),
             pdo_requested: cache_config.pdo_requested.build_with_ttl(),

src/client/node_io.rs

@@ -21,6 +21,15 @@ impl Client {
     /// a consistent (generation, Arc) pair. Must be called from a non-async
     /// context or inside a scoped block (MutexGuard is !Send).
     pub(crate) fn swap_message_semaphore(&self, permits: usize) {
+        let permits = if self.parsed_message_pre_ack_hook.get().is_some() {
+            // A pre-ACK hook can intentionally fail after Signal decrypt has
+            // advanced volatile state. Keep message processing serial while the
+            // hook is installed so pending retry/flush deferral has one inbound
+            // message to reason about at a time.
+            1
+        } else {
+            permits
+        };
         let mut guard = match self.message_processing_semaphore.lock() {
             Ok(g) => g,
             Err(poisoned) => poisoned.into_inner(),
@@ -415,9 +424,13 @@ impl Client {
         };
         match tag {
             "receipt" | "notification" | "call" => true,
-            "message" => from
-                .to_jid()
-                .is_some_and(|j| j.is_newsletter() || j.is_status_broadcast()),
+            "message" => {
+                if self.parsed_message_pre_ack_hook.get().is_some() {
+                    return false;
+                }
+                from.to_jid()
+                    .is_some_and(|j| j.is_newsletter() || j.is_status_broadcast())
+            }
             _ => false,
         }
     }

src/lib.rs

@@ -121,6 +121,7 @@ pub mod version;
 pub mod prelude {
     pub use crate::bot::{Bot, BotBuilder, BotHandle, MessageContext};
     pub use crate::client::{Client, ClientError};
+    pub use crate::message::{ParsedMessagePreAckContext, ParsedMessagePreAckHookAlreadySet};
     pub use crate::request::IqError;
     #[cfg(feature = "tokio-runtime")]
     pub use crate::runtime_impl::TokioRuntime;

src/message.rs

@@ -4,6 +4,8 @@ use crate::types::message::MessageInfo;
 use log::{debug, warn};
 use prost::Message as ProtoMessage;
 
+use std::future::Future;
+use std::pin::Pin;
 use std::sync::Arc;
 use wacore::libsignal::crypto::DecryptionError;
 use wacore::libsignal::protocol::SenderKeyDistributionMessage;
@@ -28,6 +30,61 @@ use waproto::whatsapp::{self as wa};
 /// After this many retries, we stop sending retry receipts and rely solely on PDO.
 const MAX_DECRYPT_RETRIES: u8 = 5;
 
+/// Future returned by a parsed-message pre-ACK hook.
+///
+/// The hook is awaited inline on the receive path after the inbound payload has
+/// been parsed and normalized, but before the SDK sends the delivery receipt or
+/// transport ACK. Returning an error intentionally suppresses the ACK so the
+/// server can redeliver the message later. This is useful for consumers that
+/// must durably commit an inbound message before WhatsApp considers it
+/// delivered.
+pub(crate) type ParsedMessagePreAckHookFuture =
+    Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'static>>;
+
+/// Callback registered with [`Client::set_parsed_message_pre_ack_hook`].
+///
+/// [`Client::set_parsed_message_pre_ack_hook`]: crate::Client::set_parsed_message_pre_ack_hook
+pub(crate) type ParsedMessagePreAckHook =
+    Arc<dyn Fn(ParsedMessagePreAckContext) -> ParsedMessagePreAckHookFuture + Send + Sync>;
+
+/// Message context passed to the awaited pre-ACK hook.
+///
+/// Both `message` and `info` are `Arc`-wrapped so the hook can cheaply share the
+/// same normalized values that the event bus will receive after ACK. The
+/// `client` handle is included for applications that need account or storage
+/// context while performing a durable commit.
+#[derive(Clone)]
+pub struct ParsedMessagePreAckContext {
+    pub message: Arc<wa::Message>,
+    pub info: Arc<MessageInfo>,
+    pub client: Arc<Client>,
+}
+
+impl ParsedMessagePreAckContext {
+    pub(crate) fn new(
+        message: Arc<wa::Message>,
+        info: Arc<MessageInfo>,
+        client: Arc<Client>,
+    ) -> Self {
+        Self {
+            message,
+            info,
+            client,
+        }
+    }
+}
+
+/// Returned when a parsed-message pre-ACK hook has already been registered.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
+#[error("parsed message pre-ACK hook is already registered")]
+pub struct ParsedMessagePreAckHookAlreadySet;
+
+#[derive(Clone)]
+pub(crate) struct PendingParsedMessagePreAck {
+    pub message: Arc<wa::Message>,
+    pub info: Arc<MessageInfo>,
+}
+
 /// Pre-extracted enc node payload. Holds owned copies of the fields needed for
 /// decryption so the async decrypt phase doesn't borrow the original NodeRef tree.
 pub(crate) struct EncPayload {
@@ -77,23 +134,31 @@ pub(crate) struct SessionBatchOutcome {
     duplicate: bool,
     undecryptable: bool,
     dispatched: bool,
+    pre_ack_hook_failed: bool,
     skdm_only: bool,
     plaintext_failed: bool,
     had_failure: bool,
 }
 
+#[derive(Clone, Copy, Debug, Default)]
+struct GroupBatchOutcome {
+    pre_ack_hook_failed: bool,
+}
+
 #[derive(Clone, Copy, Debug, Default)]
 struct MigrationDecryptOutcome {
     decrypted: bool,
     duplicate: bool,
     dispatched: bool,
+    pre_ack_hook_failed: bool,
     skdm_only: bool,
     plaintext_failed: bool,
 }
 
 #[derive(Clone, Copy, Debug, Default)]
 pub(crate) struct PlaintextHandleOutcome {
     dispatched: bool,
+    pre_ack_hook_failed: bool,
     skdm_only: bool,
 }