Inbound Durability Hook

use whatsapp_rust::{InboundDurabilityHook, prelude::*};
use async_trait::async_trait;
use std::sync::Arc;

struct MyStore { /* your DB connection */ }

#[async_trait]
impl InboundDurabilityHook for MyStore {
    async fn on_messages(
        &self,
        _client: Arc<Client>,
        batch: &[InboundMessage],
    ) -> anyhow::Result<()> {
        // 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(())
    }
}

let bot = Bot::builder()
    .with_backend(SqliteStore::new("whatsapp.db").await?)
    // 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 once per message, in arrival order;
        // the ack is deferred in the background.
        println!("received: {}", ctx.info.id);
    })
    .build()
    .await?;

use async_trait::async_trait;
use std::sync::Arc;
use wacore::types::events::InboundMessage;

#[async_trait]
pub trait InboundDurabilityHook {
    async fn on_messages(&self, client: Arc<Client>, batch: &[InboundMessage]) -> anyhow::Result<()>;
}

#[derive(Debug, Clone, Serialize, bon::Builder)]
#[non_exhaustive]
pub struct InboundMessage {
    pub message: Arc<wa::Message>,
    pub info: Arc<MessageInfo>,
}

// Correct idempotency key, per item in the batch
let key = (
    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 = item.info.id.clone();

cargo run --example durability_hook

#[async_trait::async_trait]
impl InboundDurabilityHook for InboxArchiver {
    async fn on_messages(
        &self,
        _client: Arc<Client>,
        batch: &[whatsapp_rust::types::events::InboundMessage],
    ) -> anyhow::Result<()> {
        // 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<CommitKey> = 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("<non-text>").replace(['\t', '\n'], " ");
                lines.push_str(&format!("{}\t{}\t{}\t{preview}\n", key.0, key.1, key.2));
                keys.push(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(())
    }
}