diff --git a/src/history_sync.rs b/src/history_sync.rs index 0fedecf86..f143ef524 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -1,26 +1,11 @@ -use crate::types::events::{Event, LazyConversation}; -use bytes::Bytes; +use crate::types::events::{Event, LazyHistorySync}; use std::sync::Arc; -use wacore::history_sync::process_history_sync; +use wacore::history_sync::{TcTokenCandidate, process_history_sync}; use wacore::store::traits::TcTokenEntry; -use wacore_binary::JidExt; use waproto::whatsapp::message::HistorySyncNotification; use crate::client::Client; -/// Partial Conversation decode — only tctoken fields, skips heavy `messages`. -#[derive(Clone, PartialEq, prost::Message)] -struct ConversationTcTokenFields { - #[prost(string, required, tag = "1")] - pub id: String, - #[prost(bytes = "vec", optional, tag = "21")] - pub tc_token: Option>, - #[prost(uint64, optional, tag = "22")] - pub tc_token_timestamp: Option, - #[prost(uint64, optional, tag = "28")] - pub tc_token_sender_timestamp: Option, -} - impl Client { pub(crate) async fn handle_history_sync( self: &Arc, @@ -71,10 +56,9 @@ impl Client { } } - /// Process history sync with streaming and lazy parsing. - /// - /// Memory efficient: raw bytes are wrapped in LazyConversation and only - /// parsed if the event handler actually accesses the conversation data. + /// Process history sync: decompress, extract internal data (tctokens, + /// pushname, nct_salt), then dispatch a single `Event::HistorySync` + /// with the full decompressed blob for on-demand consumer decoding. pub(crate) async fn process_history_sync_task( self: &Arc, message_id: String, @@ -155,103 +139,30 @@ impl Client { } }; - // Get own user for pushname extraction (moved into blocking task, no clone needed) let own_user = { let device_snapshot = self.persistence_manager.get_device_snapshot().await; device_snapshot.pn.as_ref().map(|j| j.to_non_ad().user) }; - // Check if anyone is listening for events let has_listeners = self.core.event_bus.has_handlers(); - let parse_result = if has_listeners { - // Use a bounded channel to stream raw conversation bytes as Bytes (zero-copy) - let (tx, rx) = async_channel::bounded::(4); - - // Run streaming parsing in blocking thread - // own_user is moved directly, no clone needed - let (result_tx, result_rx) = futures::channel::oneshot::channel(); - // Spawn the blocking work concurrently — it runs while we - // process channel items below. - let blocking_fut = self.runtime.spawn_blocking(Box::new(move || { - let own_user_ref = own_user.as_deref(); - - // Streaming: decompresses and extracts raw bytes incrementally - // No parsing happens here - just raw byte extraction - // Uses Bytes for zero-copy reference counting - let result = process_history_sync( - compressed_data, - own_user_ref, - Some(|raw_bytes: Bytes| { - // Send Bytes through channel (zero-copy clone) - #[cfg(not(target_arch = "wasm32"))] - let _ = tx.send_blocking(raw_bytes); - #[cfg(target_arch = "wasm32")] - let _ = tx.try_send(raw_bytes); - }), - compressed_size_hint, - ); - // tx dropped here, closing channel - let _ = result_tx.send(result); - })); - // Drive the blocking future to completion in the background - self.runtime - .spawn(Box::pin(async move { - blocking_fut.await; - })) - .detach(); - - // Receive and dispatch lazy conversations as they come in - let mut conv_count = 0usize; - while let Ok(raw_bytes) = rx.recv().await { - if self.is_shutting_down() { - log::debug!( - "Stopping history sync {} event dispatch during shutdown", - message_id - ); - break; - } - conv_count += 1; - if conv_count.is_multiple_of(25) { - log::info!("History sync progress: {conv_count} conversations processed..."); - } - // Extract tctokens before dispatching to ensure backfill even if handler drops - self.store_tc_token_from_conversation_bytes(&raw_bytes) - .await; - - // Wrap Bytes in LazyConversation using from_bytes (true zero-copy) - // Parsing only happens if the event handler calls .conversation() or .get() - let lazy_conv = LazyConversation::from_bytes(raw_bytes); - self.core.event_bus.dispatch(Event::JoinedGroup(lazy_conv)); - } - - // Drop receiver before awaiting the blocking task. If we broke out - // of the loop during shutdown, the sender may be blocked on - // tx.send_blocking() — dropping rx causes it to return Err and - // unblock, preventing a deadlock. - drop(rx); - - // Wait for parsing result - result_rx.await.ok() + // Small blobs (PushName, Recent): decode inline to avoid spawn_blocking overhead. + // Large blobs: use blocking thread to avoid stalling the async runtime. + const INLINE_THRESHOLD: usize = 256 * 1024; + let parse_result = if compressed_data.len() < INLINE_THRESHOLD { + Some(process_history_sync( + compressed_data, + own_user.as_deref(), + has_listeners, + compressed_size_hint, + )) } else { - // No event listeners, but still extract tctokens from conversations - // so headless/library clients have cached privacy tokens after pairing. - log::debug!("No event handlers registered, extracting tctokens only"); - - let (tx, rx) = async_channel::bounded::(4); - let (result_tx, result_rx) = futures::channel::oneshot::channel(); let blocking_fut = self.runtime.spawn_blocking(Box::new(move || { - let own_user_ref = own_user.as_deref(); let result = process_history_sync( compressed_data, - own_user_ref, - Some(|raw_bytes: Bytes| { - #[cfg(not(target_arch = "wasm32"))] - let _ = tx.send_blocking(raw_bytes); - #[cfg(target_arch = "wasm32")] - let _ = tx.try_send(raw_bytes); - }), + own_user.as_deref(), + has_listeners, compressed_size_hint, ); let _ = result_tx.send(result); @@ -261,16 +172,6 @@ impl Client { blocking_fut.await; })) .detach(); - - while let Ok(raw_bytes) = rx.recv().await { - if self.is_shutting_down() { - break; - } - self.store_tc_token_from_conversation_bytes(&raw_bytes) - .await; - } - drop(rx); - result_rx.await.ok() }; @@ -308,6 +209,24 @@ impl Client { ) .await; } + + // Store tctokens extracted during streaming (move to avoid cloning) + for candidate in sync_result.tc_token_candidates { + self.store_tc_token_candidate(candidate).await; + } + + // Dispatch a single event with the full decompressed blob + if let Some(decompressed) = sync_result.decompressed_bytes { + let lazy_hs = LazyHistorySync::new( + decompressed, + notification.sync_type().into(), + notification.chunk_order, + notification.progress, + ); + self.core + .event_bus + .dispatch(Event::HistorySync(Box::new(lazy_hs))); + } } Some(Err(e)) => { log::error!("Failed to process HistorySync data: {:?}", e); @@ -318,36 +237,13 @@ impl Client { } } - /// Extract and store tctoken data from a raw Conversation protobuf. - /// Partial decode — only reads fields 1/21/22/28, skipping messages. - async fn store_tc_token_from_conversation_bytes(&self, raw_bytes: &[u8]) { - use prost::Message; - - let conv = match ConversationTcTokenFields::decode(raw_bytes) { - Ok(c) => c, - Err(_) => return, - }; - - let token = match conv.tc_token { - Some(t) if !t.is_empty() => t, - _ => return, - }; - - let Some(timestamp) = conv.tc_token_timestamp else { - return; - }; - - // Resolve to LID for storage key consistency with notification handler - let jid: wacore_binary::Jid = match conv.id.parse() { + /// Store a tctoken candidate extracted during history sync streaming. + async fn store_tc_token_candidate(&self, candidate: TcTokenCandidate) { + let jid: wacore_binary::Jid = match candidate.id.parse() { Ok(j) => j, Err(_) => return, }; - // Only 1:1 conversations carry tctokens - if jid.is_group() || jid.is_newsletter() || jid.is_bot() { - return; - } - let resolved_lid = if jid.is_lid() { None } else { @@ -358,9 +254,9 @@ impl Client { let backend = self.persistence_manager.backend(); // Avoid clobbering a newer local sender_timestamp from post-send issuance - let incoming_sender_ts = conv.tc_token_sender_timestamp.map(|ts| ts as i64); + let incoming_sender_ts = candidate.tc_token_sender_timestamp.map(|ts| ts as i64); let merged_sender_ts = if let Ok(Some(existing)) = backend.get_tc_token(token_key).await { - if (existing.token_timestamp as u64) > timestamp { + if (existing.token_timestamp as u64) > candidate.tc_token_timestamp { return; } match (existing.sender_timestamp, incoming_sender_ts) { @@ -373,8 +269,8 @@ impl Client { }; let entry = TcTokenEntry { - token, - token_timestamp: timestamp as i64, + token: candidate.tc_token, + token_timestamp: candidate.tc_token_timestamp as i64, sender_timestamp: merged_sender_ts, }; @@ -389,7 +285,7 @@ impl Client { target: "Client/TcToken", "Stored tctoken from history sync for {} (t={})", token_key, - timestamp + candidate.tc_token_timestamp ); } } diff --git a/wacore/src/history_sync.rs b/wacore/src/history_sync.rs index 946527879..c511352b2 100644 --- a/wacore/src/history_sync.rs +++ b/wacore/src/history_sync.rs @@ -1,9 +1,7 @@ use bytes::Bytes; use flate2::read::ZlibDecoder; -use prost::Message; use std::io::Read; use thiserror::Error; -use waproto::whatsapp as wa; #[derive(Debug, Error)] pub enum HistorySyncError { @@ -15,7 +13,7 @@ pub enum HistorySyncError { MalformedProtobuf(String), } -#[derive(Debug, Default)] +#[derive(Debug)] pub struct HistorySyncResult { pub own_pushname: Option, /// NCT salt from HistorySync field 19 (nctSalt). @@ -23,6 +21,11 @@ pub struct HistorySyncResult { /// Source: WAWeb/History/MsgHandlerAction.js:storeNctSaltFromHistorySync pub nct_salt: Option>, pub conversations_processed: usize, + /// Tctoken candidates extracted from 1:1 conversations during streaming. + pub tc_token_candidates: Vec, + /// The full decompressed protobuf blob, only retained when event + /// listeners exist. Wrapped in `LazyHistorySync` for on-demand decoding. + pub decompressed_bytes: Option, } mod wire_type { @@ -42,36 +45,38 @@ mod wire_type { /// /// After decompression, the compressed input is dropped immediately, so peak /// memory = max(compressed, decompressed) + small overhead, not both. -pub fn process_history_sync( +pub fn process_history_sync( compressed_data: Vec, own_user: Option<&str>, - mut on_conversation_bytes: Option, + retain_blob: bool, compressed_size_hint: Option, -) -> Result -where - F: FnMut(Bytes), -{ - // Decompress into a single contiguous buffer. - // If the compressed (post-decrypt) size is known from the notification's - // file_length, use it with the 4x multiplier for a better estimate than - // guessing from the encrypted input (which includes MAC/padding overhead). +) -> Result { + // Hard limit to prevent OOM on malformed blobs. + // Typical InitialBootstrap: 5-20 MB decompressed. + const MAX_DECOMPRESSED: u64 = 64 * 1024 * 1024; + let estimated = compressed_size_hint .and_then(|s| usize::try_from(s).ok()) .map(|s| s * 4) .unwrap_or_else(|| compressed_data.len() * 4) - .clamp(256, 8 * 1024 * 1024); + .clamp(256, MAX_DECOMPRESSED as usize); let mut decompressed = Vec::with_capacity(estimated); { - let mut decoder = ZlibDecoder::new(compressed_data.as_slice()); - decoder.read_to_end(&mut decompressed)?; + let decoder = ZlibDecoder::new(compressed_data.as_slice()); + let mut limited = decoder.take(MAX_DECOMPRESSED); + limited.read_to_end(&mut decompressed)?; } - // Drop compressed data immediately — no longer needed. drop(compressed_data); - // Wrap in Bytes so we can hand out zero-copy slices. let buf = Bytes::from(decompressed); let mut pos = 0; - let mut result = HistorySyncResult::default(); + let mut result = HistorySyncResult { + own_pushname: None, + nct_salt: None, + conversations_processed: 0, + tc_token_candidates: Vec::new(), + decompressed_bytes: if retain_blob { Some(buf.clone()) } else { None }, + }; while pos < buf.len() { let (tag, bytes_read) = read_varint(&buf[pos..])?; @@ -87,16 +92,15 @@ where pos += vlen; let end = checked_end(pos, len, buf.len(), "conversation")?; - if let Some(ref mut callback) = on_conversation_bytes { - // Zero-copy slice — just an Arc refcount increment. - callback(buf.slice(pos..end)); - result.conversations_processed += 1; + result.conversations_processed += 1; + if let Some(candidate) = extract_tc_token_fields(&buf[pos..end]) { + result.tc_token_candidates.push(candidate); } pos = end; } // field 7 = pushnames (repeated, length-delimited) - 7 if own_user.is_some() + 7 if let Some(own) = own_user && result.own_pushname.is_none() && wire_type_raw == wire_type::LENGTH_DELIMITED => { @@ -104,11 +108,7 @@ where pos += vlen; let end = checked_end(pos, len, buf.len(), "pushname")?; - if let Ok(pn) = wa::Pushname::decode(&buf[pos..end]) - && let Some(ref id) = pn.id - && Some(id.as_str()) == own_user - && let Some(name) = pn.pushname - { + if let Some(name) = extract_own_pushname(&buf[pos..end], own) { result.own_pushname = Some(name); } pos = end; @@ -207,6 +207,100 @@ fn skip_field(wire_type: u32, buf: &[u8], pos: usize) -> Result Option { + let mut pos = 0; + let mut id_match = false; + let mut pushname: Option = None; + + while pos < data.len() { + let (tag, bytes_read) = read_varint(data.get(pos..)?).ok()?; + pos += bytes_read; + let field_number = (tag >> 3) as u32; + let wt = (tag & 0x7) as u32; + + match field_number { + // id (tag 1, string) + 1 if wt == wire_type::LENGTH_DELIMITED => { + let (len, vlen) = read_varint(data.get(pos..)?).ok()?; + pos += vlen; + let len = usize::try_from(len).ok()?; + let end = pos.checked_add(len).filter(|&e| e <= data.len())?; + let id = std::str::from_utf8(data.get(pos..end)?).ok()?; + id_match = id == own_user; + if !id_match { + return None; // wrong user, skip entirely + } + pos = end; + } + // pushname (tag 2, string) + 2 if wt == wire_type::LENGTH_DELIMITED => { + let (len, vlen) = read_varint(data.get(pos..)?).ok()?; + pos += vlen; + let len = usize::try_from(len).ok()?; + let end = pos.checked_add(len).filter(|&e| e <= data.len())?; + let name = std::str::from_utf8(data.get(pos..end)?).ok()?; + pushname = Some(name.to_string()); + pos = end; + } + _ => { + pos = skip_field(wt, data, pos).ok()?; + } + } + } + + if id_match { pushname } else { None } +} + +/// Prost partial decode — only tctoken fields, skips heavy `messages`. +#[derive(Clone, PartialEq, prost::Message)] +pub(crate) struct ConversationTcTokenFields { + #[prost(string, required, tag = "1")] + pub id: String, + #[prost(bytes = "vec", optional, tag = "21")] + pub tc_token: Option>, + #[prost(uint64, optional, tag = "22")] + pub tc_token_timestamp: Option, + #[prost(uint64, optional, tag = "28")] + pub tc_token_sender_timestamp: Option, +} + +/// Extract tctoken candidate from a raw Conversation proto. +/// Uses prost partial decode (only fields 1/21/22/28, skips messages). +/// Returns `None` for groups, newsletters, bots, or conversations without tctokens. +pub(crate) fn extract_tc_token_fields(data: &[u8]) -> Option { + use prost::Message; + + let conv = ConversationTcTokenFields::decode(data).ok()?; + + // Early-out for non-1:1 conversations + if let Some(parts) = wacore_binary::jid::parse_jid_fast(&conv.id) + && (parts.server == "g.us" || parts.server == "newsletter" || parts.server == "bot") + { + return None; + } + + let tc_token = conv.tc_token.filter(|t| !t.is_empty())?; + let tc_token_timestamp = conv.tc_token_timestamp?; + + Some(TcTokenCandidate { + id: conv.id, + tc_token, + tc_token_timestamp, + tc_token_sender_timestamp: conv.tc_token_sender_timestamp, + }) +} + +/// Tctoken data extracted from a conversation during streaming. +#[derive(Debug)] +pub struct TcTokenCandidate { + pub id: String, + pub tc_token: Vec, + pub tc_token_timestamp: u64, + pub tc_token_sender_timestamp: Option, +} + #[cfg(test)] mod tests { use super::*; @@ -214,6 +308,7 @@ mod tests { use flate2::write::ZlibEncoder; use prost::Message; use std::io::Write; + use waproto::whatsapp as wa; /// Encode a HistorySync proto and zlib-compress it. fn encode_and_compress(hs: &wa::HistorySync) -> Vec { @@ -233,7 +328,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync::(compressed, None, None, None).unwrap(); + let result = process_history_sync(compressed, None, false, None).unwrap(); assert_eq!(result.nct_salt, Some(salt)); } @@ -246,7 +341,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync::(compressed, None, None, None).unwrap(); + let result = process_history_sync(compressed, None, false, None).unwrap(); assert!(result.nct_salt.is_none()); } @@ -265,8 +360,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = - process_history_sync::(compressed, Some("0000000000"), None, None).unwrap(); + let result = process_history_sync(compressed, Some("0000000000"), false, None).unwrap(); assert_eq!(result.nct_salt, Some(salt)); assert_eq!(result.own_pushname.as_deref(), Some("TestUser")); diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 6f2fad501..f4ea18798 100755 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -10,122 +10,121 @@ use std::sync::{Arc, OnceLock, RwLock}; use wacore_binary::Node; use wacore_binary::OwnedNodeRef; use wacore_binary::{Jid, MessageId}; -use waproto::whatsapp::{self as wa, HistorySync}; +use waproto::whatsapp as wa; -/// A lazily-parsed conversation from history sync. +/// A lazily-parsed history sync blob. /// -/// Raw protobuf bytes are stored and only parsed on first access. -/// With `Arc` dispatch, all handlers share the same `LazyConversation` +/// Wraps the decompressed protobuf bytes and only decodes on first access. +/// With `Arc` dispatch, all handlers share the same `LazyHistorySync` /// so `OnceLock` gives parse-once semantics for free. -#[derive(Clone)] -pub struct LazyConversation { - /// Raw protobuf bytes using Bytes for zero-copy cloning. - /// Bytes is reference-counted internally, so clones share the same data. +/// +/// Cheap metadata (`sync_type`, `chunk_order`, `progress`) is available +/// without decoding — useful for filtering events. +/// +/// Call [`get()`](Self::get) for full access to conversations, pushnames, +/// global settings, past participants, call logs, and everything else in +/// the `wa::HistorySync` proto. +pub struct LazyHistorySync { raw_bytes: Bytes, - /// Cached parsed result, initialized on first access. - parsed: OnceLock, + sync_type: i32, + chunk_order: Option, + progress: Option, + parsed: OnceLock>>, } -impl LazyConversation { - /// Create a new lazy conversation from raw protobuf bytes. - /// The bytes are moved into Bytes for zero-copy sharing. - pub fn new(raw_bytes: Vec) -> Self { +impl Clone for LazyHistorySync { + fn clone(&self) -> Self { Self { - raw_bytes: Bytes::from(raw_bytes), - parsed: OnceLock::new(), + raw_bytes: self.raw_bytes.clone(), + sync_type: self.sync_type, + chunk_order: self.chunk_order, + progress: self.progress, + parsed: OnceLock::new(), // don't deep-copy the decoded proto } } +} - /// Create from an existing Bytes instance (true zero-copy). - pub fn from_bytes(raw_bytes: Bytes) -> Self { +impl LazyHistorySync { + pub fn new( + raw_bytes: Bytes, + sync_type: i32, + chunk_order: Option, + progress: Option, + ) -> Self { Self { raw_bytes, + sync_type, + chunk_order, + progress, parsed: OnceLock::new(), } } - /// Access the raw protobuf bytes for full decoding (including messages). - /// - /// Since [`get()`](Self::get) and [`conversation()`](Self::conversation) - /// strip messages to save memory, consumers that need message history - /// should decode from these bytes directly via - /// `wa::Conversation::decode(lazy_conv.raw_bytes())`. - pub fn raw_bytes(&self) -> &[u8] { - &self.raw_bytes + /// History sync type (e.g. InitialBootstrap, Recent, PushName). + /// Available without decoding the proto. + pub fn sync_type(&self) -> i32 { + self.sync_type } - /// Decode the full conversation including messages. - /// - /// Unlike [`get()`](Self::get) which strips messages to save memory, - /// this decodes a fresh copy from the raw bytes every time and keeps - /// the full `WebMessageInfo` array intact. Returns `None` if decoding - /// fails or the conversation id is empty. - /// - /// The result is not cached — call this only when you actually need - /// the messages, and prefer [`get()`](Self::get) for metadata-only access. - pub fn get_with_messages(&self) -> Option { - let conv = wa::Conversation::decode(&self.raw_bytes[..]).ok()?; - if conv.id.is_empty() { None } else { Some(conv) } + /// Chunk ordering for multi-chunk transfers. + pub fn chunk_order(&self) -> Option { + self.chunk_order } - /// Get the parsed conversation, parsing on first access. - /// Returns None if parsing fails (empty id indicates invalid conversation). - /// - /// Messages are always stripped on first parse to reduce memory — - /// history sync conversations embed full `WebMessageInfo` arrays that - /// can be very large. Use [`raw_bytes()`](Self::raw_bytes) if you need messages. - pub fn get(&self) -> Option<&wa::Conversation> { - let conv = self.parsed.get_or_init(|| { - let mut conv = wa::Conversation::decode(&self.raw_bytes[..]).unwrap_or_default(); - conv.messages.clear(); - conv.messages.shrink_to_fit(); - conv - }); - if conv.id.is_empty() { None } else { Some(conv) } + /// Sync progress (0-100). + pub fn progress(&self) -> Option { + self.progress } - /// Get the parsed conversation, parsing on first access. - /// Panics if parsing fails (use `get()` for fallible access). + /// Full decode of the history sync proto, cached via OnceLock. + /// Returns `None` if decoding fails. /// - /// Messages are always stripped on first parse to reduce memory. - pub fn conversation(&self) -> &wa::Conversation { - self.parsed.get_or_init(|| { - let mut conv = wa::Conversation::decode(&self.raw_bytes[..]) - .expect("Failed to decode conversation"); - conv.messages.clear(); - conv.messages.shrink_to_fit(); - conv - }) + /// Note: decoding materializes the full proto in memory alongside the + /// raw bytes (~2x decompressed size). For large InitialBootstrap blobs, + /// prefer [`raw_bytes()`](Self::raw_bytes) with partial decoding if + /// you only need specific fields. + pub fn get(&self) -> Option<&wa::HistorySync> { + self.parsed + .get_or_init(|| { + wa::HistorySync::decode(&self.raw_bytes[..]) + .ok() + .map(Box::new) + }) + .as_deref() + } + + /// Access the raw decompressed protobuf bytes for custom/partial decoding. + pub fn raw_bytes(&self) -> &[u8] { + &self.raw_bytes } } -impl fmt::Debug for LazyConversation { +impl fmt::Debug for LazyHistorySync { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - if let Some(conv) = self.parsed.get() { - f.debug_struct("LazyConversation") - .field("id", &conv.id) - .field("parsed", &true) - .finish() - } else { - f.debug_struct("LazyConversation") - .field("raw_size", &self.raw_bytes.len()) - .field("parsed", &false) - .finish() - } + f.debug_struct("LazyHistorySync") + .field("sync_type", &self.sync_type) + .field("chunk_order", &self.chunk_order) + .field("progress", &self.progress) + .field("raw_size", &self.raw_bytes.len()) + .field( + "parsed", + &self.parsed.get().and_then(|o| o.as_ref()).is_some(), + ) + .finish() } } -impl Serialize for LazyConversation { +impl Serialize for LazyHistorySync { fn serialize(&self, serializer: S) -> Result where S: serde::Serializer, { - // Only serialize if parsed, otherwise serialize as null/empty - if let Some(conv) = self.parsed.get() { - conv.serialize(serializer) - } else { - serializer.serialize_none() - } + use serde::ser::SerializeStruct; + let mut s = serializer.serialize_struct("LazyHistorySync", 3)?; + s.serialize_field("sync_type", &self.sync_type)?; + s.serialize_field("chunk_order", &self.chunk_order)?; + s.serialize_field("progress", &self.progress)?; + s.end() } } @@ -389,7 +388,6 @@ pub enum Event { ContactNumberChanged(ContactNumberChanged), ContactSyncRequested(ContactSyncRequested), - JoinedGroup(LazyConversation), /// Group metadata/settings/participant change from w:gp2 notification. GroupUpdate(GroupUpdate), ContactUpdate(ContactUpdate), @@ -404,7 +402,7 @@ pub enum Event { DeleteChatUpdate(DeleteChatUpdate), DeleteMessageForMeUpdate(DeleteMessageForMeUpdate), - HistorySync(HistorySync), + HistorySync(Box), OfflineSyncPreview(OfflineSyncPreview), OfflineSyncCompleted(OfflineSyncCompleted), @@ -906,129 +904,113 @@ mod tests { use prost::Message; use waproto::whatsapp as wa; - /// Build a Conversation proto with an id and N dummy messages, encode it. - fn make_conversation_bytes(id: &str, num_messages: usize) -> Vec { - let messages: Vec = (0..num_messages) - .map(|i| wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { - key: wa::MessageKey { - id: Some(format!("msg-{i}")), - ..Default::default() - }, - ..Default::default() - }), - msg_order_id: Some(i as u64), - }) - .collect(); - - let conv = wa::Conversation { - id: id.to_string(), - messages, + /// Build a HistorySync proto with conversations and encode it. + fn make_history_sync_bytes(conversations: Vec) -> Vec { + let hs = wa::HistorySync { + sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32, + conversations, ..Default::default() }; - conv.encode_to_vec() + hs.encode_to_vec() } #[test] - fn get_strips_messages() { - let bytes = make_conversation_bytes("chat@s.whatsapp.net", 5); - let lazy = LazyConversation::new(bytes); + fn lazy_history_sync_get_decodes() { + let bytes = make_history_sync_bytes(vec![wa::Conversation { + id: "chat@s.whatsapp.net".to_string(), + ..Default::default() + }]); + let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); - let conv = lazy.get().expect("should parse"); - assert_eq!(conv.id, "chat@s.whatsapp.net"); - assert!(conv.messages.is_empty(), "get() must strip messages"); + let hs = lazy.get().expect("should decode"); + assert_eq!(hs.conversations.len(), 1); + assert_eq!(hs.conversations[0].id, "chat@s.whatsapp.net"); } #[test] - fn conversation_strips_messages() { - let bytes = make_conversation_bytes("chat@s.whatsapp.net", 3); - let lazy = LazyConversation::new(bytes); - - let conv = lazy.conversation(); - assert_eq!(conv.id, "chat@s.whatsapp.net"); - assert!( - conv.messages.is_empty(), - "conversation() must strip messages" - ); + fn lazy_history_sync_caches_decode() { + let bytes = make_history_sync_bytes(vec![wa::Conversation { + id: "test@g.us".to_string(), + ..Default::default() + }]); + let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); + + let first = lazy.get().expect("first decode"); + let second = lazy.get().expect("second decode"); + // Same reference — OnceLock cached it + assert!(std::ptr::eq(first, second)); } #[test] - fn raw_bytes_returns_original_proto() { - let bytes = make_conversation_bytes("chat@s.whatsapp.net", 4); - let lazy = LazyConversation::new(bytes.clone()); + fn lazy_history_sync_cheap_metadata() { + let bytes = make_history_sync_bytes(vec![]); + let lazy = LazyHistorySync::new(Bytes::from(bytes), 3, Some(2), Some(50)); - assert_eq!(lazy.raw_bytes(), &bytes[..]); - - // Users can decode the full conversation from raw_bytes - let full = wa::Conversation::decode(lazy.raw_bytes()).expect("should decode"); - assert_eq!(full.id, "chat@s.whatsapp.net"); - assert_eq!(full.messages.len(), 4); + assert_eq!(lazy.sync_type(), 3); + assert_eq!(lazy.chunk_order(), Some(2)); + assert_eq!(lazy.progress(), Some(50)); } #[test] - fn get_with_messages_preserves_messages() { - let bytes = make_conversation_bytes("chat@s.whatsapp.net", 7); - let lazy = LazyConversation::new(bytes); + fn lazy_history_sync_raw_bytes() { + let bytes = make_history_sync_bytes(vec![wa::Conversation { + id: "raw@s.whatsapp.net".to_string(), + ..Default::default() + }]); + let raw = bytes.clone(); + let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); - let full = lazy.get_with_messages().expect("should decode"); - assert_eq!(full.id, "chat@s.whatsapp.net"); - assert_eq!(full.messages.len(), 7); - assert_eq!( - full.messages[0].message.as_ref().unwrap().key.id.as_deref(), - Some("msg-0") - ); + assert_eq!(lazy.raw_bytes(), &raw[..]); + + // Consumer can partial-decode from raw_bytes + let decoded = wa::HistorySync::decode(lazy.raw_bytes()).expect("should decode"); + assert_eq!(decoded.conversations[0].id, "raw@s.whatsapp.net"); } #[test] - fn get_with_messages_independent_of_cached_parse() { - let bytes = make_conversation_bytes("chat@s.whatsapp.net", 3); - let lazy = LazyConversation::new(bytes); - - // Trigger the cached parse first (strips messages) - let stripped = lazy.get().expect("should parse"); - assert!(stripped.messages.is_empty()); - - // get_with_messages should still return full messages - let full = lazy.get_with_messages().expect("should decode"); - assert_eq!(full.messages.len(), 3); + fn lazy_history_sync_empty_bytes_decodes_default() { + // Empty protobuf bytes are valid — decode to default HistorySync + let lazy = LazyHistorySync::new(Bytes::new(), 0, None, None); + let hs = lazy.get().expect("empty bytes decode to default"); + assert!(hs.conversations.is_empty()); } #[test] - fn get_returns_none_for_empty_id() { - let conv = wa::Conversation { - id: String::new(), - ..Default::default() - }; - let lazy = LazyConversation::new(conv.encode_to_vec()); + fn lazy_history_sync_corrupt_bytes_returns_none() { + let lazy = LazyHistorySync::new(Bytes::from_static(&[0xFF, 0xFF, 0xFF]), 0, None, None); assert!(lazy.get().is_none()); } #[test] - fn get_with_messages_returns_none_for_empty_id() { + fn lazy_history_sync_preserves_messages() { let conv = wa::Conversation { - id: String::new(), + id: "chat@s.whatsapp.net".to_string(), + messages: vec![wa::HistorySyncMsg { + message: Some(wa::WebMessageInfo { + key: wa::MessageKey { + id: Some("msg-0".to_string()), + ..Default::default() + }, + ..Default::default() + }), + msg_order_id: Some(0), + }], ..Default::default() }; - let lazy = LazyConversation::new(conv.encode_to_vec()); - assert!(lazy.get_with_messages().is_none()); - } - - #[test] - fn get_with_messages_returns_none_for_invalid_bytes() { - let lazy = LazyConversation::new(vec![0xFF, 0xFF, 0xFF]); - assert!(lazy.get_with_messages().is_none()); - } - - #[test] - fn from_bytes_works_same_as_new() { - let bytes = make_conversation_bytes("test@s.whatsapp.net", 2); - let lazy = LazyConversation::from_bytes(Bytes::from(bytes)); - - let full = lazy.get_with_messages().expect("should decode"); - assert_eq!(full.id, "test@s.whatsapp.net"); - assert_eq!(full.messages.len(), 2); + let bytes = make_history_sync_bytes(vec![conv]); + let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); - let stripped = lazy.get().expect("should parse"); - assert!(stripped.messages.is_empty()); + let hs = lazy.get().expect("should decode"); + assert_eq!(hs.conversations[0].messages.len(), 1); + assert_eq!( + hs.conversations[0].messages[0] + .message + .as_ref() + .unwrap() + .key + .id + .as_deref(), + Some("msg-0") + ); } }