Custom Backends - WhatsApp Rust

Overview

This guide covers implementing custom backends for storage, network transport, and HTTP operations in whatsapp-rust. The library uses trait-based abstractions to allow full customization.

Storage backend architecture

The storage backend is split into four domain-grouped traits:

SignalStore - Signal protocol cryptography (identities, sessions, keys)
AppSyncStore - WhatsApp app state synchronization
ProtocolStore - WhatsApp Web protocol alignment (SKDM, LID mapping, device registry)
DeviceStore - Device persistence operations

See Store API reference for the complete trait definitions.

Backend Trait

Any type implementing all four traits automatically implements Backend:

use wacore::store::traits::*;

pub trait Backend: SignalStore + AppSyncStore + ProtocolStore + DeviceStore + Send + Sync {}

impl<T> Backend for T 
where 
    T: SignalStore + AppSyncStore + ProtocolStore + DeviceStore + Send + Sync 
{}

Implementing a custom storage backend

Step 1: define your store

use std::sync::Arc;
use async_trait::async_trait;
use wacore::store::traits::*;
use wacore::store::error::{Result, StoreError};

#[derive(Clone)]
pub struct MyCustomStore {
    // Your storage implementation (database connection, file handle, etc.)
    connection: Arc<MyConnection>,
}

impl MyCustomStore {
    pub async fn new(connection_string: &str) -> Result<Self> {
        let connection = MyConnection::connect(connection_string).await?;
        Ok(Self {
            connection: Arc::new(connection),
        })
    }
}

Step 2: implement SignalStore

Handle Signal protocol cryptographic operations:

use bytes::Bytes;

#[async_trait]
impl SignalStore for MyCustomStore {
    // --- Identity Operations ---
    
    async fn put_identity(&self, address: &str, key: [u8; 32]) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO identities (address, key) VALUES (?, ?)",
            &[address, &key[..]],
        ).await?;
        Ok(())
    }
    
    async fn load_identity(&self, address: &str) -> Result<Option<[u8; 32]>> {
        let row = self.connection.query_optional(
            "SELECT key FROM identities WHERE address = ?",
            &[address],
        ).await?;
        Ok(row.map(|r| {
            let bytes: Vec<u8> = r.get(0);
            let mut arr = [0u8; 32];
            arr.copy_from_slice(&bytes);
            arr
        }))
    }
    
    async fn delete_identity(&self, address: &str) -> Result<()> {
        self.connection.execute(
            "DELETE FROM identities WHERE address = ?",
            &[address],
        ).await?;
        Ok(())
    }
    
    // --- Session Operations ---
    
    async fn get_session(&self, address: &str) -> Result<Option<Bytes>> {
        let row = self.connection.query_optional(
            "SELECT record FROM sessions WHERE address = ?",
            &[address],
        ).await?;
        Ok(row.map(|r| Bytes::from(r.get::<Vec<u8>>(0))))
    }
    
    async fn put_session(&self, address: &str, session: &[u8]) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO sessions (address, record) VALUES (?, ?)",
            &[address, session],
        ).await?;
        Ok(())
    }
    
    async fn delete_session(&self, address: &str) -> Result<()> {
        self.connection.execute(
            "DELETE FROM sessions WHERE address = ?",
            &[address],
        ).await?;
        Ok(())
    }
    
    // --- PreKey Operations ---
    
    async fn store_prekey(&self, id: u32, record: &[u8], uploaded: bool) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO prekeys (id, record, uploaded) VALUES (?, ?, ?)",
            &[&id, record, &(uploaded as i32)],
        ).await?;
        Ok(())
    }
    
    async fn load_prekey(&self, id: u32) -> Result<Option<Bytes>> {
        let row = self.connection.query_optional(
            "SELECT record FROM prekeys WHERE id = ?",
            &[&id],
        ).await?;
        Ok(row.map(|r| Bytes::from(r.get::<Vec<u8>>(0))))
    }

    // Override load_prekeys_batch for better performance (default loops over load_prekey)
    async fn load_prekeys_batch(&self, ids: &[u32]) -> Result<Vec<(u32, Bytes)>> {
        let placeholders = ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let query = format!("SELECT id, record FROM prekeys WHERE id IN ({})", placeholders);
        let rows = self.connection.query_all(&query, ids).await?;
        Ok(rows.iter().map(|r| (r.get(0), Bytes::from(r.get::<Vec<u8>>(1)))).collect())
    }
    
    async fn remove_prekey(&self, id: u32) -> Result<()> {
        self.connection.execute(
            "DELETE FROM prekeys WHERE id = ?",
            &[&id],
        ).await?;
        Ok(())
    }
    
    async fn get_max_prekey_id(&self) -> Result<u32> {
        let row = self.connection.query_optional(
            "SELECT MAX(id) FROM prekeys",
            &[],
        ).await?;
        Ok(row.and_then(|r| r.get(0)).unwrap_or(0))
    }
    
    // --- Signed PreKey Operations ---
    
    async fn store_signed_prekey(&self, id: u32, record: &[u8]) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO signed_prekeys (id, record) VALUES (?, ?)",
            &[&id, record],
        ).await?;
        Ok(())
    }
    
    async fn load_signed_prekey(&self, id: u32) -> Result<Option<Vec<u8>>> {
        let row = self.connection.query_optional(
            "SELECT record FROM signed_prekeys WHERE id = ?",
            &[&id],
        ).await?;
        Ok(row.map(|r| r.get(0)))
    }
    
    async fn load_all_signed_prekeys(&self) -> Result<Vec<(u32, Vec<u8>)>> {
        let rows = self.connection.query(
            "SELECT id, record FROM signed_prekeys",
            &[],
        ).await?;
        Ok(rows.into_iter().map(|r| (r.get(0), r.get(1))).collect())
    }
    
    async fn remove_signed_prekey(&self, id: u32) -> Result<()> {
        self.connection.execute(
            "DELETE FROM signed_prekeys WHERE id = ?",
            &[&id],
        ).await?;
        Ok(())
    }
    
    // --- Sender Key Operations ---
    
    async fn put_sender_key(&self, address: &str, record: &[u8]) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO sender_keys (address, record) VALUES (?, ?)",
            &[address, record],
        ).await?;
        Ok(())
    }
    
    async fn get_sender_key(&self, address: &str) -> Result<Option<Vec<u8>>> {
        let row = self.connection.query_optional(
            "SELECT record FROM sender_keys WHERE address = ?",
            &[address],
        ).await?;
        Ok(row.map(|r| r.get(0)))
    }
    
    async fn delete_sender_key(&self, address: &str) -> Result<()> {
        self.connection.execute(
            "DELETE FROM sender_keys WHERE address = ?",
            &[address],
        ).await?;
        Ok(())
    }
}

See Store API reference for all SignalStore methods.

Step 3: implement AppSyncStore

Handle WhatsApp app state synchronization:

use wacore::appstate::hash::HashState;
use wacore_appstate::processor::AppStateMutationMAC;

#[async_trait]
impl AppSyncStore for MyCustomStore {
    async fn get_sync_key(&self, key_id: &[u8]) -> Result<Option<AppStateSyncKey>> {
        let row = self.connection.query_optional(
            "SELECT key_data, fingerprint, timestamp FROM app_state_sync_keys WHERE key_id = ?",
            &[key_id],
        ).await?;
        
        Ok(row.map(|r| AppStateSyncKey {
            key_data: r.get(0),
            fingerprint: r.get(1),
            timestamp: r.get(2),
        }))
    }
    
    async fn set_sync_key(&self, key_id: &[u8], key: AppStateSyncKey) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO app_state_sync_keys (key_id, key_data, fingerprint, timestamp) VALUES (?, ?, ?, ?)",
            &[key_id, &key.key_data[..], &key.fingerprint[..], &key.timestamp],
        ).await?;
        Ok(())
    }
    
    async fn get_version(&self, name: &str) -> Result<HashState> {
        let row = self.connection.query_optional(
            "SELECT version, hash FROM app_state_versions WHERE name = ?",
            &[name],
        ).await?;
        
        Ok(row.map(|r| HashState {
            version: r.get(0),
            hash: r.get(1),
        }).unwrap_or_default())
    }
    
    async fn set_version(&self, name: &str, state: HashState) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO app_state_versions (name, version, hash) VALUES (?, ?, ?)",
            &[name, &state.version, &state.hash[..]],
        ).await?;
        Ok(())
    }
    
    async fn put_mutation_macs(
        &self,
        name: &str,
        version: u64,
        mutations: &[AppStateMutationMAC],
    ) -> Result<()> {
        for mutation in mutations {
            self.connection.execute(
                "INSERT INTO mutation_macs (name, version, index_mac, value_mac) VALUES (?, ?, ?, ?)",
                &[name, &version, &mutation.index_mac[..], &mutation.value_mac[..]],
            ).await?;
        }
        Ok(())
    }
    
    async fn get_mutation_mac(&self, name: &str, index_mac: &[u8]) -> Result<Option<Vec<u8>>> {
        let row = self.connection.query_optional(
            "SELECT value_mac FROM mutation_macs WHERE name = ? AND index_mac = ?",
            &[name, index_mac],
        ).await?;
        Ok(row.map(|r| r.get(0)))
    }
    
    async fn delete_mutation_macs(&self, name: &str, index_macs: &[Vec<u8>]) -> Result<()> {
        for index_mac in index_macs {
            self.connection.execute(
                "DELETE FROM mutation_macs WHERE name = ? AND index_mac = ?",
                &[name, &index_mac[..]],
            ).await?;
        }
        Ok(())
    }
    
    async fn get_latest_sync_key_id(&self) -> Result<Option<Vec<u8>>> {
        let row = self.connection.query_optional(
            "SELECT key_id FROM app_state_sync_keys ORDER BY timestamp DESC LIMIT 1",
            &[],
        ).await?;
        Ok(row.map(|r| r.get(0)))
    }
}

See Store API reference for all AppSyncStore methods.

Step 4: implement ProtocolStore

Handle WhatsApp Web protocol alignment:

use wacore_binary::jid::Jid;

#[async_trait]
impl ProtocolStore for MyCustomStore {
    // --- Per-Device Sender Key Tracking ---
    
    async fn get_sender_key_devices(&self, group_jid: &str) -> Result<Vec<(String, bool)>> {
        let rows = self.connection.query(
            "SELECT device_jid, has_key FROM sender_key_devices WHERE group_jid = ?",
            &[group_jid],
        ).await?;
        
        Ok(rows.into_iter()
            .map(|r| (r.get::<String>(0), r.get::<i32>(1) != 0))
            .collect())
    }
    
    async fn set_sender_key_status(&self, group_jid: &str, entries: &[(&str, bool)]) -> Result<()> {
        for (device_jid, has_key) in entries {
            self.connection.execute(
                "INSERT OR REPLACE INTO sender_key_devices (group_jid, device_jid, has_key, updated_at) VALUES (?, ?, ?, strftime('%s', 'now'))",
                &[group_jid, device_jid, &(*has_key as i32)],
            ).await?;
        }
        Ok(())
    }
    
    async fn clear_sender_key_devices(&self, group_jid: &str) -> Result<()> {
        self.connection.execute(
            "DELETE FROM sender_key_devices WHERE group_jid = ?",
            &[group_jid],
        ).await?;
        Ok(())
    }
    
    // --- LID-PN Mapping ---
    
    async fn get_lid_mapping(&self, lid: &str) -> Result<Option<LidPnMappingEntry>> {
        let row = self.connection.query_optional(
            "SELECT lid, phone_number, created_at, updated_at, learning_source FROM lid_pn_mappings WHERE lid = ?",
            &[lid],
        ).await?;
        
        Ok(row.map(|r| LidPnMappingEntry {
            lid: r.get(0),
            phone_number: r.get(1),
            created_at: r.get(2),
            updated_at: r.get(3),
            learning_source: r.get(4),
        }))
    }
    
    async fn get_pn_mapping(&self, phone: &str) -> Result<Option<LidPnMappingEntry>> {
        let row = self.connection.query_optional(
            "SELECT lid, phone_number, created_at, updated_at, learning_source FROM lid_pn_mappings WHERE phone_number = ? ORDER BY updated_at DESC LIMIT 1",
            &[phone],
        ).await?;
        
        Ok(row.map(|r| LidPnMappingEntry {
            lid: r.get(0),
            phone_number: r.get(1),
            created_at: r.get(2),
            updated_at: r.get(3),
            learning_source: r.get(4),
        }))
    }
    
    async fn put_lid_mapping(&self, entry: &LidPnMappingEntry) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO lid_pn_mappings (lid, phone_number, created_at, updated_at, learning_source) VALUES (?, ?, ?, ?, ?)",
            &[&entry.lid, &entry.phone_number, &entry.created_at, &entry.updated_at, &entry.learning_source],
        ).await?;
        Ok(())
    }
    
    async fn get_all_lid_mappings(&self) -> Result<Vec<LidPnMappingEntry>> {
        let rows = self.connection.query(
            "SELECT lid, phone_number, created_at, updated_at, learning_source FROM lid_pn_mappings",
            &[],
        ).await?;
        
        Ok(rows.into_iter().map(|r| LidPnMappingEntry {
            lid: r.get(0),
            phone_number: r.get(1),
            created_at: r.get(2),
            updated_at: r.get(3),
            learning_source: r.get(4),
        }).collect())
    }
    
    // --- Sent Message Store (retry support) ---

    async fn store_sent_message(
        &self,
        chat_jid: &str,
        message_id: &str,
        payload: &[u8],
    ) -> Result<()> {
        self.connection.execute(
            "INSERT OR REPLACE INTO sent_messages (chat_jid, message_id, payload, created_at) VALUES (?, ?, ?, strftime('%s', 'now'))",
            &[chat_jid, message_id, payload],
        ).await?;
        Ok(())
    }

    async fn take_sent_message(
        &self,
        chat_jid: &str,
        message_id: &str,
    ) -> Result<Option<Vec<u8>>> {
        // Atomic read-and-delete
        let row = self.connection.query_optional(
            "DELETE FROM sent_messages WHERE chat_jid = ? AND message_id = ? RETURNING payload",
            &[chat_jid, message_id],
        ).await?;
        Ok(row.map(|r| r.get(0)))
    }

    async fn delete_expired_sent_messages(&self, cutoff_timestamp: i64) -> Result<u32> {
        let count = self.connection.execute(
            "DELETE FROM sent_messages WHERE created_at < ?",
            &[&cutoff_timestamp],
        ).await?;
        Ok(count as u32)
    }

    // ... Implement remaining methods (base_key, device_list, tc_token)
}

See Store API reference for all ProtocolStore methods.

Pending Inbound Buffer

If you want to support the opt-in InboundDurabilityHook, 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 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:

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.

#[async_trait]
impl ProtocolStore for MyCustomStore {
    // ... other methods ...

    // --- Pending Inbound Buffer (required for InboundDurabilityHook) ---
    //
    // 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,
        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) \
             VALUES (?, ?, ?, ?, ?, strftime('%s', 'now'))",
            &[chat, sender, id, &device_id, message],
        ).await?;
        Ok(())
    }

    async fn get_pending_inbound(
        &self,
        chat: &str,
        sender: &str,
        id: &str,
    ) -> Result<Option<Vec<u8>>> {
        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 = ?",
            &[chat, sender, id, &device_id],
        ).await?;
        Ok(row.map(|r| r.get(0)))
    }

    async fn delete_pending_inbound(
        &self,
        chat: &str,
        sender: &str,
        id: &str,
    ) -> Result<()> {
        let device_id = self.device_id;
        self.connection.execute(
            "DELETE FROM pending_inbound_messages \
             WHERE chat = ? AND sender = ? AND id = ? AND device_id = ?",
            &[chat, sender, id, &device_id],
        ).await?;
        Ok(())
    }

    // 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, cutoff_timestamp: i64) -> Result<u32> {
        let device_id = self.device_id;
        let count = self.connection.execute(
            "DELETE FROM pending_inbound_messages \
             WHERE device_id = ? AND inserted_at < ?",
            &[&device_id, &cutoff_timestamp],
        ).await?;
        Ok(count as u32)
    }
}

The recommended schema (matching the bundled SQLite migration):

CREATE TABLE pending_inbound_messages (
    chat        TEXT    NOT NULL,
    sender      TEXT    NOT NULL,
    id          TEXT    NOT NULL,
    message     BLOB    NOT NULL,
    device_id   INTEGER NOT NULL DEFAULT 1,
    inserted_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')),
    PRIMARY KEY (chat, sender, id, device_id)
);

-- For the per-device retention sweep (delete_expired_pending_inbound).
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.

Step 5: implement DeviceStore

Handle device data persistence:

use wacore::store::Device;

#[async_trait]
impl DeviceStore for MyCustomStore {
    async fn save(&self, device: &Device) -> Result<()> {
        // Serialize device data (use your preferred format)
        let serialized = bincode::serialize(device)
            .map_err(|e| StoreError::Serialization(Box::new(e)))?;
        
        self.connection.execute(
            "UPDATE devices SET data = ? WHERE id = 1",
            &[&serialized[..]],
        ).await?;
        Ok(())
    }
    
    async fn load(&self) -> Result<Option<Device>> {
        let row = self.connection.query_optional(
            "SELECT data FROM devices WHERE id = 1",
            &[],
        ).await?;
        
        match row {
            Some(r) => {
                let data: Vec<u8> = r.get(0);
                let device = bincode::deserialize(&data)
                    .map_err(|e| StoreError::Serialization(Box::new(e)))?;
                Ok(Some(device))
            }
            None => Ok(None),
        }
    }
    
    async fn exists(&self) -> Result<bool> {
        let row = self.connection.query_optional(
            "SELECT 1 FROM devices WHERE id = 1",
            &[],
        ).await?;
        Ok(row.is_some())
    }
    
    async fn create(&self) -> Result<i32> {
        self.connection.execute(
            "INSERT INTO devices (id, data) VALUES (1, ?)",
            &[&Vec::<u8>::new()],
        ).await?;
        Ok(1)
    }
    
    async fn snapshot_db(&self, name: &str, extra_content: Option<&[u8]>) -> Result<()> {
        // Optional: Implement database snapshotting for debugging
        Ok(())
    }
}

See Store API reference for all DeviceStore methods.

Using your custom backend

use whatsapp_rust::bot::Bot;
use whatsapp_rust::TokioRuntime;

let backend = Arc::new(MyCustomStore::new("my://connection").await?);

let bot = Bot::builder()
    .with_backend(backend)
    .with_transport_factory(transport_factory)
    .with_http_client(http_client)
    .with_runtime(TokioRuntime)
    .build()
    .await?;

Custom Runtime

The Runtime trait is the foundation of the runtime-agnostic architecture. It abstracts async task spawning, sleeping, blocking operations, and cooperative yielding. Implement it to use a different async runtime (e.g., async-std, smol, or a WASM executor).

The Runtime trait

use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
use async_trait::async_trait;
use wacore::runtime::{AbortHandle, Runtime};

pub struct MyRuntime;

#[async_trait]
impl Runtime for MyRuntime {
    fn spawn(
        &self,
        future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>,
    ) -> AbortHandle {
        // Spawn a task on your runtime and return an AbortHandle
        // that cancels the task when called
        let handle = my_runtime::spawn(future);
        AbortHandle::new(move || handle.cancel())
    }

    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send>> {
        // Return a future that completes after the given duration
        Box::pin(my_runtime::sleep(duration))
    }

    fn spawn_blocking(
        &self,
        f: Box<dyn FnOnce() + Send + 'static>,
    ) -> Pin<Box<dyn Future<Output = ()> + Send>> {
        // Offload a blocking closure to a thread pool
        Box::pin(async {
            let _ = my_runtime::spawn_blocking(f).await;
        })
    }

    fn yield_now(&self) -> Option<Pin<Box<dyn Future<Output = ()> + Send>>> {
        // Return Some(future) if cooperative yielding is needed,
        // or None if the runtime handles this automatically
        // (e.g., multi-threaded runtimes don't need explicit yielding)
        None
    }

    fn yield_frequency(&self) -> u32 {
        // How many items to process before yielding in tight loops.
        // Default is 10. Single-threaded runtimes should return 1
        // to avoid starving the event loop.
        10
    }
}

AbortHandle

AbortHandle is a type-erased cancellation handle marked #[must_use]. Dropping an AbortHandle aborts the spawned task, ensuring cleanup. Call .detach() to prevent automatic cancellation for fire-and-forget tasks where the task should run to completion even if the parent scope is dropped.

Because AbortHandle is #[must_use], the compiler warns if you discard the return value of Runtime::spawn(). You must either store the handle (to abort later), call .detach() (to let the task run freely), or explicitly call .abort().

// Store the handle to abort later
let handle = runtime.spawn(Box::pin(my_task()));
// ...
handle.abort(); // Cancel the task

// Detach for fire-and-forget tasks
runtime.spawn(Box::pin(background_work())).detach();

As of #1124, Runtime also has spawn_detached for fire-and-forget spawns:

runtime.spawn_detached(Box::pin(background_work()));

The trait-level default is exactly self.spawn(future).detach() — the same cost as writing both calls out by hand, kept only so existing Runtime implementations keep compiling and behaving unchanged. It is not itself the optimization: spawn still boxes a dyn FnOnce capturing the executor’s own handle to build the AbortHandle, and the default still builds and immediately drops it. The saving requires overriding spawn_detached to bypass spawn entirely. The bundled TokioRuntime does exactly that in src/runtime_impl.rs:

fn spawn_detached(&self, future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
    tokio::spawn(future);
}

This never constructs an AbortHandle at all. Implement spawn_detached this way for your own runtime if it can spawn a task with no cancellation bookkeeping; otherwise the inherited default is a correct (if not faster) fallback.

Using your runtime

use whatsapp_rust::bot::Bot;

let mut bot = Bot::builder()
    .with_backend(backend)
    .with_transport_factory(transport_factory)
    .with_http_client(http_client)
    .with_runtime(MyRuntime)
    .build()
    .await?;

To target WASM, disable all default features and implement Runtime without Send bounds. On target_arch = "wasm32", the trait automatically removes Send requirements.

Spawnable trait (v0.6). The future-bound that generic spawn helpers use is now the Spawnable marker trait instead of a hardcoded Send + 'static. On native targets Spawnable = Send + 'static; on wasm32 it’s just 'static (WASM runtimes are single-threaded and can’t require Send). Runtime::spawn’s signature is unchanged — this only matters if you write your own generic helpers over spawnable futures, in which case bound them on wacore::runtime::Spawnable rather than Send + 'static so they compile on both targets.

If you are implementing a single-threaded runtime, override yield_frequency() to return 1 and have yield_now() return Some(future). This ensures the event loop is not starved during tight processing loops (e.g., decoding incoming frames).

Proxy and custom TLS

whatsapp-rust has two separate network layers, each with its own extension point for proxy and TLS customization:

Layer	Purpose	Extension point
WebSocket transport	WhatsApp protocol connection	`TransportFactory` trait / `with_connector`
HTTP client	Media upload/download, version fetching	`HttpClient` trait / `UreqHttpClient::with_agent`

To route all traffic through a proxy, you need to configure both layers.

Custom TLS for the WebSocket transport

Use TokioWebSocketTransportFactory::with_connector to supply a custom TLS Connector — for example, adding custom CA certificates or client certificates:

use whatsapp_rust_tokio_transport::{TokioWebSocketTransportFactory, Connector};
use tokio_rustls::TlsConnector;
use rustls::RootCertStore;
use std::sync::Arc;

let mut root_store = RootCertStore::empty();
root_store.add_parsable_certificates(my_ca_certs);

let tls_config = rustls::ClientConfig::builder()
    .with_root_certificates(root_store)
    .with_no_client_auth();

let connector = Connector::Rustls(TlsConnector::from(Arc::new(tls_config)));

let factory = TokioWebSocketTransportFactory::new()
    .with_connector(connector);

Use default_tls_connector() to inspect or replicate the default TLS configuration as a starting point.

Custom Origin for a relay

TokioWebSocketTransportFactory sends Origin: https://web.whatsapp.com on every upgrade by default, matching what whatsmeow and Baileys send unconditionally. If with_url points at a relay or gateway that enforces its own Origin check, override it with with_origin:

let factory = TokioWebSocketTransportFactory::new()
    .with_url("wss://my-relay.example/ws/chat")
    .with_origin("https://my-relay.example");

Keep the default for a relay that simply forwards traffic to WhatsApp — it’s still emulating a real WA Web browser, so Origin should stay https://web.whatsapp.com regardless of which host you dial. Only call with_origin when the relay itself checks the header against its own domain. without_origin() drops the header entirely, for a peer that rejects the upgrade because of it. See Transport API reference for details.

Full proxy support for the WebSocket transport

For full proxy support (SOCKS5, HTTP CONNECT, etc.), implement the TransportFactory trait directly. Establish the WebSocket connection through your proxy, then wrap it with from_websocket:

use whatsapp_rust_tokio_transport::from_websocket;
use wacore::net::{Transport, TransportEvent, TransportFactory};
use async_trait::async_trait;
use std::sync::Arc;

struct ProxyTransportFactory {
    proxy_addr: String,
    target_url: String,
}

#[async_trait]
impl TransportFactory for ProxyTransportFactory {
    async fn create_transport(
        &self,
    ) -> Result<(Arc<dyn Transport>, async_channel::Receiver<TransportEvent>), anyhow::Error> {
        // 1. Connect to proxy (example: SOCKS5)
        let proxy_stream = tokio_socks::tcp::Socks5Stream::connect(
            &self.proxy_addr,
            "web.whatsapp.com:443",
        ).await?;

        // 2. Upgrade to TLS + WebSocket through the proxy
        let (ws, _) = tokio_websockets::ClientBuilder::from_uri(
            self.target_url.parse()?,
        )
        .connect_on(proxy_stream)
        .await?;

        // 3. Wrap the WebSocket into a Transport + event channel
        Ok(from_websocket(ws))
    }
}

Proxy support for the HTTP client

Use UreqHttpClient::with_agent to supply a pre-configured ureq::Agent with proxy settings:

use whatsapp_rust_ureq_http_client::UreqHttpClient;
use ureq::config::Config;

let agent: ureq::Agent = Config::builder()
    .proxy(ureq::Proxy::new("socks5://127.0.0.1:1080")?)
    .build()
    .into();

let http_client = UreqHttpClient::with_agent(agent);

Alternatively, implement the HttpClient trait directly for full control over HTTP behavior.

Combining both layers

To route all traffic through a proxy, configure both the transport factory and the HTTP client:

use whatsapp_rust::bot::Bot;
use whatsapp_rust::TokioRuntime;

let mut bot = Bot::builder()
    .with_backend(backend)
    .with_transport_factory(ProxyTransportFactory {
        proxy_addr: "127.0.0.1:1080".to_string(),
        target_url: "wss://web.whatsapp.com/ws/chat".to_string(),
    })
    .with_http_client(UreqHttpClient::with_agent(proxy_agent))
    .with_runtime(TokioRuntime)
    .on_event(|event, client| async move { /* ... */ })
    .build()
    .await?;

The WebSocket transport handles the persistent WhatsApp protocol connection, while the HTTP client handles media operations (upload/download) and version fetching. Both must be configured separately for full proxy coverage.

Custom transport backend

Transport Trait

Implement the Transport trait for custom network transports:

use async_trait::async_trait;
use wacore::net::Transport;

pub struct MyCustomTransport {
    // Your transport implementation
}

#[async_trait]
impl Transport for MyCustomTransport {
    async fn send(&self, data: Vec<u8>) -> Result<(), anyhow::Error> {
        // Send raw bytes through your transport
        Ok(())
    }
    
    async fn disconnect(&self) {
        // Close the connection
    }
}

See Transport API reference for full trait details.

Transport Factory

Implement TransportFactory to create transport instances:

use wacore::net::{TransportFactory, TransportEvent};
use async_channel::Receiver;
use std::sync::Arc;

pub struct MyCustomTransportFactory {
    url: String,
}

impl MyCustomTransportFactory {
    pub fn new(url: String) -> Self {
        Self { url }
    }
}

#[async_trait]
impl TransportFactory for MyCustomTransportFactory {
    async fn create_transport(
        &self,
    ) -> Result<(Arc<dyn Transport>, Receiver<TransportEvent>), anyhow::Error> {
        // Create transport and event channel
        let (event_tx, event_rx) = async_channel::bounded(10000);
        
        // Connect to server
        let transport = MyCustomTransport::connect(&self.url).await?;
        
        // Spawn read pump task
        tokio::spawn(read_pump(transport.reader(), event_tx));
        
        // Send Connected event
        event_tx.send(TransportEvent::Connected).await?;
        
        Ok((Arc::new(transport), event_rx))
    }
}

See Transport API reference for factory details.

Transport Events

use wacore::net::DisconnectReason;

// TransportEvent (from wacore::net) — shown here for reference.
pub enum TransportEvent {
    Connected,
    DataReceived(Bytes),
    Disconnected(DisconnectReason),
}

// Read pump example
async fn read_pump(
    mut reader: impl AsyncRead,
    event_tx: async_channel::Sender<TransportEvent>,
) {
    loop {
        match read_frame(&mut reader).await {
            Ok(data) => {
                if event_tx.send(TransportEvent::DataReceived(data)).await.is_err() {
                    break;
                }
            }
            Err(_) => break,
        }
    }
    let _ = event_tx.send(TransportEvent::Disconnected(DisconnectReason::StreamEnded)).await;
}

Custom HTTP Client

Implement the HttpClient trait for custom HTTP operations:

Make self.send_request(&request) below return Ok for a completed exchange, even a 4xx/5xx status. Reserve Err for two cases: the exchange never starts at all (DNS failure, connection refused, a TLS error, a timeout), or a body read fails after a 2xx status (for example, the body exceeds a size cap you enforce).Media download and upload read status_code off the Ok response to refresh a stale media-auth token (401/403) or move to the next CDN host on other errors like 5xx; download additionally re-derives an expired URL (404/410) the same way. If you turn a non-2xx status into Err instead, you hide it from all of this, and every retry repeats the same stale token. See HTTP Client Trait for the full contract, including how the bundled UreqHttpClient works around ureq’s opposite default.

use whatsapp_rust::http::{HttpClient, HttpRequest, HttpResponse};
use wacore::net::StreamingHttpResponse;
use async_trait::async_trait;

pub struct MyCustomHttpClient {
    // Your HTTP client implementation
}

#[async_trait]
impl HttpClient for MyCustomHttpClient {
    async fn execute(&self, request: HttpRequest) -> Result<HttpResponse, anyhow::Error> {
        // Execute HTTP request
        let response = self.send_request(&request).await?;
        
        Ok(HttpResponse {
            status_code: response.status_code,
            body: response.body,
        })
    }

    // Override supports_streaming if you implement execute_streaming.
    // When false (the default), download_to_writer falls back to buffered downloads.
    fn supports_streaming(&self) -> bool {
        true
    }
    
    fn execute_streaming(&self, request: HttpRequest) -> Result<StreamingHttpResponse, anyhow::Error> {
        // Return a streaming response with a reader
        Ok(StreamingHttpResponse {
            status_code: response.status_code,
            body: Box::new(response.body_reader),
        })
    }
}

If your HTTP client doesn’t support streaming, you only need to implement execute. The supports_streaming method defaults to false, and download_to_writer will automatically use a buffered fallback.

SQLite reference implementation

The library includes a full SQLite implementation you can use as reference: See Store API reference for the SQLite implementation details.

Key Features

Diesel ORM with migrations
Connection pooling (r2d2)
WAL mode for concurrency
Prepared statements for performance
Transaction support
Database snapshotting for debugging

Using SQLite store

use whatsapp_rust::store::SqliteStore;

let backend = Arc::new(SqliteStore::new("whatsapp.db").await?);

SQLite is bundled by default via the bundled-sqlite feature on whatsapp-rust-sqlite-storage. No system SQLite installation is required.

Custom cache store

The pluggable cache store adapter lets you replace the default in-process caches with an external backend like Redis or Memcached. This is useful for sharing cache state across multiple client instances or for deployments where in-process memory is limited.

The CacheStore trait

Implement the CacheStore trait from wacore::store::cache:

use async_trait::async_trait;
use std::time::Duration;
// You can import from either location:
// use wacore::store::cache::CacheStore;
use whatsapp_rust::CacheStore;

pub struct MyRedisCacheStore {
    client: redis::aio::ConnectionManager,
}

impl MyRedisCacheStore {
    pub async fn new(url: &str) -> Self {
        let client = redis::Client::open(url).unwrap();
        let conn = client.get_connection_manager().await.unwrap();
        Self { client: conn }
    }
}

#[async_trait]
impl CacheStore for MyRedisCacheStore {
    async fn get(&self, namespace: &str, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
        let redis_key = format!("{namespace}:{key}");
        let result: Option<Vec<u8>> = redis::cmd("GET")
            .arg(&redis_key)
            .query_async(&mut self.client.clone())
            .await?;
        Ok(result)
    }

    async fn set(
        &self,
        namespace: &str,
        key: &str,
        value: &[u8],
        ttl: Option<Duration>,
    ) -> anyhow::Result<()> {
        let redis_key = format!("{namespace}:{key}");
        match ttl {
            Some(duration) => {
                redis::cmd("SET")
                    .arg(&redis_key)
                    .arg(value)
                    .arg("EX")
                    .arg(duration.as_secs())
                    .query_async(&mut self.client.clone())
                    .await?;
            }
            None => {
                redis::cmd("SET")
                    .arg(&redis_key)
                    .arg(value)
                    .query_async(&mut self.client.clone())
                    .await?;
            }
        }
        Ok(())
    }

    async fn delete(&self, namespace: &str, key: &str) -> anyhow::Result<()> {
        let redis_key = format!("{namespace}:{key}");
        redis::cmd("DEL")
            .arg(&redis_key)
            .query_async(&mut self.client.clone())
            .await?;
        Ok(())
    }

    async fn clear(&self, namespace: &str) -> anyhow::Result<()> {
        // Use SCAN to find and delete all keys with the namespace prefix
        let pattern = format!("{namespace}:*");
        let mut cursor = 0u64;
        loop {
            let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
                .arg(cursor)
                .arg("MATCH")
                .arg(&pattern)
                .arg("COUNT")
                .arg(100)
                .query_async(&mut self.client.clone())
                .await?;

            if !keys.is_empty() {
                redis::cmd("DEL")
                    .arg(&keys)
                    .query_async(&mut self.client.clone())
                    .await?;
            }

            cursor = next_cursor;
            if cursor == 0 {
                break;
            }
        }
        Ok(())
    }

    async fn entry_count(&self, namespace: &str) -> anyhow::Result<u64> {
        // Approximate count using SCAN (expensive — use only for diagnostics)
        let pattern = format!("{namespace}:*");
        let mut count = 0u64;
        let mut cursor = 0u64;
        loop {
            let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
                .arg(cursor)
                .arg("MATCH")
                .arg(&pattern)
                .arg("COUNT")
                .arg(100)
                .query_async(&mut self.client.clone())
                .await?;
            count += keys.len() as u64;
            cursor = next_cursor;
            if cursor == 0 {
                break;
            }
        }
        Ok(count)
    }
}

Plugging in your cache store

Use CacheStores to assign your custom backend to specific caches:

use whatsapp_rust::{CacheConfig, CacheStores};
use std::sync::Arc;

let redis = Arc::new(MyRedisCacheStore::new("redis://localhost:6379").await);

// Route only group and device caches to Redis
let bot = Bot::builder()
    .with_backend(backend)
    .with_transport_factory(transport)
    .with_http_client(http_client)
    .with_runtime(TokioRuntime)
    .with_cache_config(CacheConfig {
        cache_stores: CacheStores {
            group_cache: Some(redis.clone()),
            device_registry_cache: Some(redis.clone()),
            ..Default::default()
        },
        ..Default::default()
    })
    .build()
    .await?;

Or route all pluggable caches to the same backend:

let config = CacheConfig {
    cache_stores: CacheStores::all(redis.clone()),
    ..Default::default()
};

Available namespaces

The following namespaces are used internally by the client:

Namespace	Cache	Description
`"group"`	`group_cache`	Group metadata
`"device_registry"`	`device_registry_cache`	Device registry entries
`"lid_pn_by_lid"`	`lid_pn_cache`	LID-to-phone bidirectional mappings

Design considerations

Error handling is best-effort. Cache misses and failures are logged as warnings but don’t break the client — it falls back to fetching from the authoritative source.
Serialization uses serde_json. Values are serialized to JSON bytes on the custom-store path. The in-process path has zero serialization overhead.
TTL is forwarded from CacheEntryConfig. Your implementation receives the same TTL configured in CacheConfig.
Coordination caches cannot be externalized. Session locks, message queues, and enqueue locks hold live Rust objects (mutexes, channels) and always stay in-process.
invalidate_all() requires tokio-runtime. The synchronous invalidate_all() method on TypedCache spawns a fire-and-forget task via Tokio for custom backends. Without the tokio-runtime feature, the clear is skipped with a warning. Use the async clear() method instead if you disable tokio-runtime.

Best Practices

Next Steps

Sending Messages - Use your custom backend
Receiving Messages - Store received messages
Group Management - Store group metadata
Inbound Durability Hook - At-least-once delivery with your backend

​Overview

​Storage backend architecture

​Backend Trait

​Implementing a custom storage backend

​Step 1: define your store

​Step 2: implement SignalStore

​Step 3: implement AppSyncStore

​Step 4: implement ProtocolStore

​Pending Inbound Buffer

​Step 5: implement DeviceStore

​Using your custom backend

​Custom Runtime

​The Runtime trait

​AbortHandle

​Using your runtime

​Proxy and custom TLS

​Custom TLS for the WebSocket transport

​Custom Origin for a relay

​Full proxy support for the WebSocket transport

​Proxy support for the HTTP client

​Combining both layers

​Custom transport backend

​Transport Trait

​Transport Factory

​Transport Events

​Custom HTTP Client

​SQLite reference implementation

​Key Features

​Using SQLite store

​Custom cache store

​The CacheStore trait

​Plugging in your cache store

​Available namespaces

​Design considerations

​Best Practices

​Next Steps

Overview

Storage backend architecture

Backend Trait

Implementing a custom storage backend

Step 1: define your store

Step 2: implement SignalStore

Step 3: implement AppSyncStore

Step 4: implement ProtocolStore

Pending Inbound Buffer

Step 5: implement DeviceStore

Using your custom backend

Custom Runtime

The Runtime trait

AbortHandle

Using your runtime

Proxy and custom TLS

Custom TLS for the WebSocket transport

Custom Origin for a relay

Full proxy support for the WebSocket transport

Proxy support for the HTTP client

Combining both layers

Custom transport backend

Transport Trait

Transport Factory

Transport Events

Custom HTTP Client

SQLite reference implementation

Key Features

Using SQLite store

Custom cache store

The CacheStore trait

Plugging in your cache store

Available namespaces

Design considerations

Best Practices

Next Steps