From fc21156f0a1f8c9fb1239839a473a18f6340354f Mon Sep 17 00:00:00 2001 From: Jean-Louis Queguiner Date: Thu, 10 Sep 2026 20:09:49 -0400 Subject: [PATCH 1/3] fix(whatsapp): store history sync delivered as JoinedGroup events wa-rs 0.2 never dispatches `Event::HistorySync`. It streams the backfill one conversation at a time through `Event::JoinedGroup(LazyConversation)` (see wa-rs `history_sync.rs`: "Receive and dispatch lazy conversations as they come in"). The variant still exists in the enum, so the arm matching it kept compiling while receiving nothing, and every conversation WhatsApp pushed after pairing was dropped. Measured on a fresh link: wa-rs logged "History sync progress: 775 conversations processed" while only 4 rows reached the database, and the handler's own log line never appeared once. Split the per-conversation body out of `handle_history_sync` into `store_conversation` and call it from the `JoinedGroup` arm, so history is persisted as it streams in. The `HistorySync` arm is kept: it costs nothing and resumes working if wa-rs dispatches it again. Progress is reported as a cumulative counter every 250 messages rather than one line per conversation, since a backfill carries hundreds. --- CHANGELOG.md | 2 + .../src/connector/connector_trait.rs | 28 +++++++++++- crates/void-whatsapp/src/connector/sync.rs | 43 +++++++++++++++---- 3 files changed, 63 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 019b223..b632f92 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **WhatsApp** — History sync is stored again. `wa-rs` 0.2 streams the backfill one conversation at a time through `Event::JoinedGroup(LazyConversation)` and never dispatches `Event::HistorySync`, so the handler listening for the latter was dead code and every conversation WhatsApp sent after pairing was dropped. Observed on a fresh link: 775 conversations parsed by `wa-rs`, 4 rows stored. Conversations are now persisted as they arrive, with a cumulative progress line every 250 messages. + - **Archive** — `void archive ` now dismisses the whole context group behind the item (Slack thread, Slack 1-hour channel group, Gmail thread) instead of a single row. The inbox shows one row per context, so archiving only the visible id let an older sibling resurface as the next representative. The response gains `archived_count` (rows newly archived by the call, `0` when it was already archived), and Gmail pushes the group in one `batchModify` request. ### Added diff --git a/crates/void-whatsapp/src/connector/connector_trait.rs b/crates/void-whatsapp/src/connector/connector_trait.rs index e8eaaf9..6615a2f 100644 --- a/crates/void-whatsapp/src/connector/connector_trait.rs +++ b/crates/void-whatsapp/src/connector/connector_trait.rs @@ -1,3 +1,4 @@ +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use async_trait::async_trait; @@ -15,7 +16,7 @@ use void_core::models::*; use crate::CONNECTOR_ID; use super::presence::schedule_unavailable; -use super::sync::{handle_history_sync, handle_message, render_qr}; +use super::sync::{handle_history_sync, handle_message, render_qr, store_conversation}; use super::WhatsAppConnector; #[async_trait] @@ -93,6 +94,9 @@ impl Connector for WhatsAppConnector { let config_id = self.config_id.clone(); let client_holder = Arc::clone(&self.client); let own_identity_holder = Arc::clone(&self.own_identity); + // Compteur cumule de messages d'historique importes, partage entre les + // appels du handler (un par conversation pendant un backfill). + let history_count = Arc::new(AtomicU64::new(0)); let mut bot = Bot::builder() .with_backend(backend) @@ -103,6 +107,7 @@ impl Connector for WhatsAppConnector { let config_id = config_id.clone(); let client_holder = Arc::clone(&client_holder); let own_identity_holder = Arc::clone(&own_identity_holder); + let history_count = Arc::clone(&history_count); async move { { let mut holder = client_holder.lock().await; @@ -180,6 +185,27 @@ impl Connector for WhatsAppConnector { "WhatsApp mute update ignored (mute list is managed in config.toml)" ); } + Event::JoinedGroup(lazy_conv) => { + // wa-rs 0.2 delivers history sync here, one + // conversation per event, not through + // Event::HistorySync (never dispatched). + let own_identity = own_identity_holder.lock().expect("mutex").clone(); + let conv = lazy_conv.conversation(); + match store_conversation(&db, &config_id, &own_identity, conv) { + Ok(0) => {} + Ok(n) => { + let hist = history_count.fetch_add(n, Ordering::Relaxed) + n; + // Cumulative counter rather than one line per + // conversation: a backfill carries hundreds. + if hist % 250 < n { + eprintln!( + "[whatsapp:{config_id}] history sync: {hist} messages imported" + ); + } + } + Err(e) => warn!("Failed to store history conversation: {e}"), + } + } Event::HistorySync(history) => { let own_identity = own_identity_holder.lock().expect("mutex").clone(); let sync_type = history.sync_type; diff --git a/crates/void-whatsapp/src/connector/sync.rs b/crates/void-whatsapp/src/connector/sync.rs index a87caa1..410b608 100644 --- a/crates/void-whatsapp/src/connector/sync.rs +++ b/crates/void-whatsapp/src/connector/sync.rs @@ -4,7 +4,7 @@ use tracing::{debug, info}; use wa_rs::proto_helpers::MessageExt; use wa_rs::types::message::MessageInfo; -use wa_rs_proto::whatsapp::{HistorySync, Message as WaMessage}; +use wa_rs_proto::whatsapp::{Conversation as WaConversation, HistorySync, Message as WaMessage}; use void_core::db::Database; use void_core::models::*; @@ -34,9 +34,40 @@ pub(super) fn handle_history_sync( let mut total_stored = 0u64; for conv in &history.conversations { + total_stored += store_conversation(db, connection_id, own_identity, conv)?; + } + + info!( + connection_id = %connection_id, + sync_type = history.sync_type, + stored = total_stored, + "history sync processed" + ); + Ok(()) +} + +/// Stores one history conversation and its messages. Returns the number stored. +/// +/// Split out of `handle_history_sync`: `wa-rs` 0.2 never dispatches +/// `Event::HistorySync`. It streams the backfill one conversation at a time as +/// `Event::JoinedGroup(LazyConversation)` (see `history_sync.rs`, "Receive and +/// dispatch lazy conversations as they come in"). The `Event::HistorySync` +/// variant still exists in the enum, so the arm matching it kept compiling +/// while silently receiving nothing. +/// +/// Measured on a fresh pairing before the fix: 775 conversations parsed by +/// `wa-rs`, 4 rows stored, and not a single "history sync" line in the log. +pub(super) fn store_conversation( + db: &Database, + connection_id: &str, + own_identity: &OwnIdentity, + conv: &WaConversation, +) -> anyhow::Result { + let mut total_stored = 0u64; + { let chat_jid = &conv.id; if chat_jid.is_empty() { - continue; + return Ok(0); } let is_group = chat_jid.ends_with("@g.us"); let conv_id = format!("wa_{connection_id}_{chat_jid}"); @@ -172,13 +203,7 @@ pub(super) fn handle_history_sync( } } - info!( - connection_id = %connection_id, - sync_type = history.sync_type, - stored = total_stored, - "history sync processed" - ); - Ok(()) + Ok(total_stored) } pub(super) struct StoredMessageInfo { From a8202f9593f8add76308f1750aad1a26ad73ce25 Mon Sep 17 00:00:00 2001 From: Jean-Louis Queguiner Date: Fri, 11 Sep 2026 15:38:30 -0400 Subject: [PATCH 2/3] fix(whatsapp): decode history conversations with get(), not conversation() Review catch on #74: the JoinedGroup arm decoded the backfill with LazyConversation::conversation(), which clears conv.messages after decoding as a memory optimisation (wa-rs-core 0.2, types/events.rs:87-96). The handler then stored conversation metadata and zero messages, so the fix was functionally identical to the broken state it replaced, with a green CI. Switch to LazyConversation::get(), which keeps the messages and returns None on a malformed payload instead of panicking. Also from the review: - comment on the history counter was in French, now English like the rest - drop the vestigial braces left in store_conversation after the extraction Tests pin the trap so it cannot come back silently: one asserts get() keeps the messages while conversation() empties them on the same payload, one asserts store_conversation persists 3 messages from get(), one asserts it persists 0 from conversation(). --- Cargo.lock | 1 + crates/void-whatsapp/Cargo.toml | 3 + .../src/connector/connector_trait.rs | 36 ++- crates/void-whatsapp/src/connector/sync.rs | 252 +++++++++--------- crates/void-whatsapp/src/connector/tests.rs | 106 ++++++++ 5 files changed, 257 insertions(+), 141 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d772798..6cbbc21 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4963,6 +4963,7 @@ dependencies = [ "async-trait", "base64 0.23.1", "chrono", + "prost", "qr2term", "serde", "serde_json", diff --git a/crates/void-whatsapp/Cargo.toml b/crates/void-whatsapp/Cargo.toml index 2b1a6ca..5c73235 100644 --- a/crates/void-whatsapp/Cargo.toml +++ b/crates/void-whatsapp/Cargo.toml @@ -29,3 +29,6 @@ chrono = { workspace = true } [dev-dependencies] uuid = { workspace = true } +# Encodes protobuf fixtures for the history-sync tests. Must track the prost +# version wa-rs-proto is built against. +prost = "0.14" diff --git a/crates/void-whatsapp/src/connector/connector_trait.rs b/crates/void-whatsapp/src/connector/connector_trait.rs index 6615a2f..7ce2e35 100644 --- a/crates/void-whatsapp/src/connector/connector_trait.rs +++ b/crates/void-whatsapp/src/connector/connector_trait.rs @@ -94,8 +94,8 @@ impl Connector for WhatsAppConnector { let config_id = self.config_id.clone(); let client_holder = Arc::clone(&self.client); let own_identity_holder = Arc::clone(&self.own_identity); - // Compteur cumule de messages d'historique importes, partage entre les - // appels du handler (un par conversation pendant un backfill). + // Cumulative counter of imported history messages, shared across handler + // calls (one per conversation during a backfill). let history_count = Arc::new(AtomicU64::new(0)); let mut bot = Bot::builder() @@ -189,21 +189,29 @@ impl Connector for WhatsAppConnector { // wa-rs 0.2 delivers history sync here, one // conversation per event, not through // Event::HistorySync (never dispatched). + // + // Use get(), not conversation(): the latter clears + // conv.messages after decoding to save memory, so it + // would hand us metadata with an empty message list. + // get() keeps the messages and returns None on a + // malformed payload instead of panicking. let own_identity = own_identity_holder.lock().expect("mutex").clone(); - let conv = lazy_conv.conversation(); - match store_conversation(&db, &config_id, &own_identity, conv) { - Ok(0) => {} - Ok(n) => { - let hist = history_count.fetch_add(n, Ordering::Relaxed) + n; - // Cumulative counter rather than one line per - // conversation: a backfill carries hundreds. - if hist % 250 < n { - eprintln!( - "[whatsapp:{config_id}] history sync: {hist} messages imported" - ); + if let Some(conv) = lazy_conv.get() { + match store_conversation(&db, &config_id, &own_identity, conv) { + Ok(0) => {} + Ok(n) => { + let hist = + history_count.fetch_add(n, Ordering::Relaxed) + n; + // Cumulative counter rather than one line per + // conversation: a backfill carries hundreds. + if hist % 250 < n { + eprintln!( + "[whatsapp:{config_id}] history sync: {hist} messages imported" + ); + } } + Err(e) => warn!("Failed to store history conversation: {e}"), } - Err(e) => warn!("Failed to store history conversation: {e}"), } } Event::HistorySync(history) => { diff --git a/crates/void-whatsapp/src/connector/sync.rs b/crates/void-whatsapp/src/connector/sync.rs index 410b608..b3ee4e7 100644 --- a/crates/void-whatsapp/src/connector/sync.rs +++ b/crates/void-whatsapp/src/connector/sync.rs @@ -64,143 +64,141 @@ pub(super) fn store_conversation( conv: &WaConversation, ) -> anyhow::Result { let mut total_stored = 0u64; - { - let chat_jid = &conv.id; - if chat_jid.is_empty() { - return Ok(0); - } - let is_group = chat_jid.ends_with("@g.us"); - let conv_id = format!("wa_{connection_id}_{chat_jid}"); - - let last_ts = conv - .messages - .iter() - .filter_map(|m| m.message.as_ref()?.message_timestamp) - .max() - .map(|t| t as i64); - - let conv_name = conv.name.clone().unwrap_or_else(|| chat_jid.clone()); - let is_self = own_identity.is_self_chat(chat_jid); - let conversation = Conversation { - id: conv_id.clone(), - connection_id: connection_id.to_string(), - connector: "whatsapp".into(), - external_id: chat_jid.clone(), - name: Some(if is_self { - SELF_CHAT_DISPLAY_NAME.to_string() - } else { - conv_name - }), - kind: if is_group { - ConversationKind::Group - } else if is_self { - ConversationKind::SelfChat - } else { - ConversationKind::Dm - }, - last_message_at: last_ts, - unread_count: conv.unread_count.unwrap_or(0) as i64, - is_muted: false, - metadata: None, - }; - db.upsert_conversation(&conversation)?; - - let mut sorted_msgs: Vec<_> = conv - .messages - .iter() - .filter_map(|m| { - let wmi = m.message.as_ref()?; - let wa_msg = wmi.message.as_ref()?; - let ts = wmi.message_timestamp? as i64; - let key = &wmi.key; - let msg_id = key.id.as_deref().unwrap_or_default(); - if msg_id.is_empty() { - return None; - } - Some((wmi, wa_msg, ts, msg_id)) - }) - .collect(); - sorted_msgs.sort_by_key(|&(_, _, ts, _)| ts); + let chat_jid = &conv.id; + if chat_jid.is_empty() { + return Ok(0); + } + let is_group = chat_jid.ends_with("@g.us"); + let conv_id = format!("wa_{connection_id}_{chat_jid}"); - let mut prev_context_id: Option = None; - let mut prev_ts: Option = None; + let last_ts = conv + .messages + .iter() + .filter_map(|m| m.message.as_ref()?.message_timestamp) + .max() + .map(|t| t as i64); + + let conv_name = conv.name.clone().unwrap_or_else(|| chat_jid.clone()); + let is_self = own_identity.is_self_chat(chat_jid); + let conversation = Conversation { + id: conv_id.clone(), + connection_id: connection_id.to_string(), + connector: "whatsapp".into(), + external_id: chat_jid.clone(), + name: Some(if is_self { + SELF_CHAT_DISPLAY_NAME.to_string() + } else { + conv_name + }), + kind: if is_group { + ConversationKind::Group + } else if is_self { + ConversationKind::SelfChat + } else { + ConversationKind::Dm + }, + last_message_at: last_ts, + unread_count: conv.unread_count.unwrap_or(0) as i64, + is_muted: false, + metadata: None, + }; + db.upsert_conversation(&conversation)?; - for (wmi, wa_msg, msg_ts, msg_id) in &sorted_msgs { - if is_system_message(wa_msg) { - continue; + let mut sorted_msgs: Vec<_> = conv + .messages + .iter() + .filter_map(|m| { + let wmi = m.message.as_ref()?; + let wa_msg = wmi.message.as_ref()?; + let ts = wmi.message_timestamp? as i64; + let key = &wmi.key; + let msg_id = key.id.as_deref().unwrap_or_default(); + if msg_id.is_empty() { + return None; } + Some((wmi, wa_msg, ts, msg_id)) + }) + .collect(); + sorted_msgs.sort_by_key(|&(_, _, ts, _)| ts); - let body = extract_text(wa_msg); - let media_type = extract_media_type(wa_msg); - let media_metadata = extract_media_metadata(wa_msg); + let mut prev_context_id: Option = None; + let mut prev_ts: Option = None; - if body.is_none() && media_type.is_none() { - continue; - } + for (wmi, wa_msg, msg_ts, msg_id) in &sorted_msgs { + if is_system_message(wa_msg) { + continue; + } - let from_me = wmi.key.from_me.unwrap_or(false); - let sender_jid = if from_me { - own_identity - .lid_jid - .clone() - .or_else(|| own_identity.phone_jid.clone()) - .or_else(|| { - wmi.key - .participant - .clone() - .or_else(|| wmi.participant.clone()) - }) - .unwrap_or_else(|| connection_id.to_string()) - } else if is_group { - wmi.key - .participant - .clone() - .or_else(|| wmi.participant.clone()) - .unwrap_or_else(|| chat_jid.clone()) - } else { - chat_jid.clone() - }; + let body = extract_text(wa_msg); + let media_type = extract_media_type(wa_msg); + let media_metadata = extract_media_metadata(wa_msg); + + if body.is_none() && media_type.is_none() { + continue; + } + + let from_me = wmi.key.from_me.unwrap_or(false); + let sender_jid = if from_me { + own_identity + .lid_jid + .clone() + .or_else(|| own_identity.phone_jid.clone()) + .or_else(|| { + wmi.key + .participant + .clone() + .or_else(|| wmi.participant.clone()) + }) + .unwrap_or_else(|| connection_id.to_string()) + } else if is_group { + wmi.key + .participant + .clone() + .or_else(|| wmi.participant.clone()) + .unwrap_or_else(|| chat_jid.clone()) + } else { + chat_jid.clone() + }; - let sender_name = wmi.push_name.clone(); + let sender_name = wmi.push_name.clone(); - let context_id = if let (Some(prev_cid), Some(pt)) = (&prev_context_id, prev_ts) { - if (*msg_ts - pt).abs() <= 3600 { - prev_cid.clone() - } else { - format!("wa_{connection_id}-group-{chat_jid}-{msg_ts}") - } + let context_id = if let (Some(prev_cid), Some(pt)) = (&prev_context_id, prev_ts) { + if (*msg_ts - pt).abs() <= 3600 { + prev_cid.clone() } else { format!("wa_{connection_id}-group-{chat_jid}-{msg_ts}") - }; - - prev_context_id = Some(context_id.clone()); - prev_ts = Some(*msg_ts); - - let reply_to_id = extract_quoted_id(wa_msg); - - let message = void_core::models::Message { - id: format!("wa_{connection_id}_{msg_id}"), - conversation_id: conv_id.clone(), - connection_id: connection_id.to_string(), - connector: "whatsapp".into(), - external_id: msg_id.to_string(), - sender: sender_jid, - sender_name, - sender_avatar_url: None, - body, - timestamp: *msg_ts, - synced_at: None, - is_archived: false, - is_saved: false, - reply_to_id, - media_type, - metadata: media_metadata, - context_id: Some(context_id), - context: None, - }; - db.upsert_message(&message)?; - total_stored += 1; - } + } + } else { + format!("wa_{connection_id}-group-{chat_jid}-{msg_ts}") + }; + + prev_context_id = Some(context_id.clone()); + prev_ts = Some(*msg_ts); + + let reply_to_id = extract_quoted_id(wa_msg); + + let message = void_core::models::Message { + id: format!("wa_{connection_id}_{msg_id}"), + conversation_id: conv_id.clone(), + connection_id: connection_id.to_string(), + connector: "whatsapp".into(), + external_id: msg_id.to_string(), + sender: sender_jid, + sender_name, + sender_avatar_url: None, + body, + timestamp: *msg_ts, + synced_at: None, + is_archived: false, + is_saved: false, + reply_to_id, + media_type, + metadata: media_metadata, + context_id: Some(context_id), + context: None, + }; + db.upsert_message(&message)?; + total_stored += 1; } Ok(total_stored) diff --git a/crates/void-whatsapp/src/connector/tests.rs b/crates/void-whatsapp/src/connector/tests.rs index e55a8fd..5cac91d 100644 --- a/crates/void-whatsapp/src/connector/tests.rs +++ b/crates/void-whatsapp/src/connector/tests.rs @@ -799,3 +799,109 @@ fn is_system_message_pin_in_chat() { }; assert!(sync::is_system_message(&msg)); } + +// --- history sync via Event::JoinedGroup ------------------------------------- +// +// Regression guard for PR #74 review: `LazyConversation::conversation()` clears +// `conv.messages` after decoding as a memory optimisation, so storing from it +// persists conversation metadata and zero messages. The connector must decode +// with `get()`, which keeps the messages. These tests pin that difference so a +// future edit back to `conversation()` fails loudly instead of silently +// dropping the backfill. + +fn history_conversation_bytes(chat_jid: &str, texts: &[&str]) -> Vec { + use prost::Message as _; + use wa_rs_proto::whatsapp::{ + Conversation as WaConversation, HistorySyncMsg, MessageKey, WebMessageInfo, + }; + + let messages = texts + .iter() + .enumerate() + .map(|(i, text)| HistorySyncMsg { + message: Some(WebMessageInfo { + key: MessageKey { + remote_jid: Some(chat_jid.to_string()), + from_me: Some(false), + id: Some(format!("MSG{i}")), + ..Default::default() + }, + message: Some(WaMessage { + conversation: Some((*text).to_string()), + ..Default::default() + }), + message_timestamp: Some(1_700_000_000 + i as u64), + push_name: Some("Tester".into()), + ..Default::default() + }), + msg_order_id: Some(i as u64), + }) + .collect(); + + let conv = WaConversation { + id: chat_jid.to_string(), + name: Some("History chat".into()), + messages, + ..Default::default() + }; + conv.encode_to_vec() +} + +#[test] +fn lazy_conversation_conversation_strips_messages_but_get_keeps_them() { + use wa_rs::types::events::LazyConversation; + + let bytes = history_conversation_bytes("33612345678@s.whatsapp.net", &["one", "two"]); + + // get(): messages preserved. This is what the connector relies on. + let lazy = LazyConversation::new(bytes.clone()); + let conv = lazy.get().expect("valid conversation"); + assert_eq!(conv.messages.len(), 2); + + // conversation(): same payload, messages cleared by the memory optimisation. + let lazy = LazyConversation::new(bytes); + assert!(lazy.conversation().messages.is_empty()); +} + +#[test] +fn lazy_conversation_get_returns_none_on_garbage() { + use wa_rs::types::events::LazyConversation; + + // Empty payload decodes to a default Conversation with an empty id. + // get() reports None instead of panicking like conversation() would. + assert!(LazyConversation::new(Vec::new()).get().is_none()); +} + +#[test] +fn store_conversation_from_lazy_get_persists_messages() { + use wa_rs::types::events::LazyConversation; + + let db = void_core::db::Database::open_in_memory().unwrap(); + let bytes = history_conversation_bytes("33612345678@s.whatsapp.net", &["one", "two", "three"]); + let lazy = LazyConversation::new(bytes); + let own = OwnIdentity::default(); + + let stored = sync::store_conversation(&db, "test-conn", &own, lazy.get().unwrap()).unwrap(); + + assert_eq!(stored, 3); + let rows = db + .list_messages("wa_test-conn_33612345678@s.whatsapp.net", 10, None, None) + .unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!(rows[0].body.as_deref(), Some("one")); +} + +#[test] +fn store_conversation_from_lazy_conversation_stores_nothing() { + use wa_rs::types::events::LazyConversation; + + // The bug this PR review caught: metadata lands, messages do not. + let db = void_core::db::Database::open_in_memory().unwrap(); + let bytes = history_conversation_bytes("33698765432@s.whatsapp.net", &["one", "two"]); + let lazy = LazyConversation::new(bytes); + let own = OwnIdentity::default(); + + let stored = sync::store_conversation(&db, "test-conn", &own, lazy.conversation()).unwrap(); + + assert_eq!(stored, 0); +} From d5020c84d26c57839b2de6adde844b214a20d67e Mon Sep 17 00:00:00 2001 From: Maxime Gaudin Date: Sun, 13 Sep 2026 09:39:21 +0200 Subject: [PATCH 3/3] fix(whatsapp): warn on skipped history conversations and serialize backfill A dropped payload used to vanish with no log, and wa-rs spawn-per-event decoded hundreds of conversations at once. Surface the skip and decode one conversation at a time. Co-authored-by: Cursor --- CHANGELOG.md | 2 +- .../src/connector/connector_trait.rs | 16 ++++++++++++++-- crates/void-whatsapp/src/connector/tests.rs | 6 ++++-- 3 files changed, 19 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b632f92..3bf4c78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- **WhatsApp** — History sync is stored again. `wa-rs` 0.2 streams the backfill one conversation at a time through `Event::JoinedGroup(LazyConversation)` and never dispatches `Event::HistorySync`, so the handler listening for the latter was dead code and every conversation WhatsApp sent after pairing was dropped. Observed on a fresh link: 775 conversations parsed by `wa-rs`, 4 rows stored. Conversations are now persisted as they arrive, with a cumulative progress line every 250 messages. +- **WhatsApp** — History sync after pairing is stored again. The library now delivers the backfill one conversation at a time, so the old bulk handler never ran and the pairing dump was dropped (observed: 775 conversations parsed, 4 rows stored). Progress is logged every 250 messages. - **Archive** — `void archive ` now dismisses the whole context group behind the item (Slack thread, Slack 1-hour channel group, Gmail thread) instead of a single row. The inbox shows one row per context, so archiving only the visible id let an older sibling resurface as the next representative. The response gains `archived_count` (rows newly archived by the call, `0` when it was already archived), and Gmail pushes the group in one `batchModify` request. diff --git a/crates/void-whatsapp/src/connector/connector_trait.rs b/crates/void-whatsapp/src/connector/connector_trait.rs index 7ce2e35..a7ac118 100644 --- a/crates/void-whatsapp/src/connector/connector_trait.rs +++ b/crates/void-whatsapp/src/connector/connector_trait.rs @@ -97,6 +97,10 @@ impl Connector for WhatsAppConnector { // Cumulative counter of imported history messages, shared across handler // calls (one per conversation during a backfill). let history_count = Arc::new(AtomicU64::new(0)); + // wa-rs Bot spawns one tokio task per event. Serialize decode+store so + // a hundreds-of-conversations backfill does not decode every payload + // at once while they convoy on Database's mutex. + let history_gate = Arc::new(tokio::sync::Mutex::new(())); let mut bot = Bot::builder() .with_backend(backend) @@ -108,6 +112,7 @@ impl Connector for WhatsAppConnector { let client_holder = Arc::clone(&client_holder); let own_identity_holder = Arc::clone(&own_identity_holder); let history_count = Arc::clone(&history_count); + let history_gate = Arc::clone(&history_gate); async move { { let mut holder = client_holder.lock().await; @@ -193,8 +198,10 @@ impl Connector for WhatsAppConnector { // Use get(), not conversation(): the latter clears // conv.messages after decoding to save memory, so it // would hand us metadata with an empty message list. - // get() keeps the messages and returns None on a - // malformed payload instead of panicking. + // get() keeps the messages and returns None when + // decode yields an empty id (empty or undecodable + // payload). + let _guard = history_gate.lock().await; let own_identity = own_identity_holder.lock().expect("mutex").clone(); if let Some(conv) = lazy_conv.get() { match store_conversation(&db, &config_id, &own_identity, conv) { @@ -212,6 +219,11 @@ impl Connector for WhatsAppConnector { } Err(e) => warn!("Failed to store history conversation: {e}"), } + } else { + warn!( + connection_id = %config_id, + "skipping history conversation with empty or undecodable id" + ); } } Event::HistorySync(history) => { diff --git a/crates/void-whatsapp/src/connector/tests.rs b/crates/void-whatsapp/src/connector/tests.rs index 5cac91d..90eaf82 100644 --- a/crates/void-whatsapp/src/connector/tests.rs +++ b/crates/void-whatsapp/src/connector/tests.rs @@ -867,9 +867,11 @@ fn lazy_conversation_conversation_strips_messages_but_get_keeps_them() { fn lazy_conversation_get_returns_none_on_garbage() { use wa_rs::types::events::LazyConversation; - // Empty payload decodes to a default Conversation with an empty id. - // get() reports None instead of panicking like conversation() would. + // Empty payload decodes to a default Conversation with an empty id, not a + // panic: prost succeeds on [] and get() is None because id is empty. assert!(LazyConversation::new(Vec::new()).get().is_none()); + let empty = LazyConversation::new(Vec::new()); + assert!(empty.conversation().id.is_empty()); } #[test]