perf(history-sync)!: store the compressed payload and expose a streaming reader

src/client/messaging.rs

@@ -425,10 +425,10 @@ fn build_secret_message_edit(
             ),
             remote_key_id: None,
         }),
-        message_context_info: Some(wa::MessageContextInfo {
+        message_context_info: Some(Box::new(wa::MessageContextInfo {
             message_secret: Some(message_secret.to_vec()),
             ..Default::default()
-        }),
+        })),
         ..Default::default()
     })
 }

src/features/comments.rs

@@ -103,10 +103,10 @@ impl<'a> Comments<'a> {
                 enc_payload: Some(enc_payload),
                 enc_iv: Some(iv.to_vec()),
             }),
-            message_context_info: Some(wa::MessageContextInfo {
+            message_context_info: Some(Box::new(wa::MessageContextInfo {
                 message_secret: Some(comment_secret.clone()),
                 ..Default::default()
-            }),
+            })),
             ..Default::default()
         };
         let result = client.send_message(chat, message).await?;

src/features/events.rs

@@ -58,10 +58,10 @@ impl<'a> Events<'a> {
             rand::make_rng::<rand::rngs::StdRng>().fill_bytes(&mut secret);
             secret
         };
-        message.message_context_info = Some(wa::MessageContextInfo {
+        message.message_context_info = Some(Box::new(wa::MessageContextInfo {
             message_secret: Some(message_secret.clone()),
             ..Default::default()
-        });
+        }));
 
         let result = self.client.send_message(to, message).await?;
         Ok((result, message_secret))

src/features/polls.rs

@@ -89,10 +89,10 @@ impl<'a> Polls<'a> {
             secret
         };
 
-        message.message_context_info = Some(wa::MessageContextInfo {
+        message.message_context_info = Some(Box::new(wa::MessageContextInfo {
             message_secret: Some(message_secret.clone()),
             ..Default::default()
-        });
+        }));
 
         let result = self.client.send_message(to, message).await?;
         Ok((result, message_secret))

src/history_sync.rs

@@ -1,4 +1,4 @@
-use crate::types::events::{Event, EventKind, LazyHistorySync};
+use crate::types::events::{Event, LazyHistorySync};
 use std::sync::Arc;
 use wacore::history_sync::{HistoryMsgSecretRecord, TcTokenCandidate, process_history_sync};
 use wacore::store::traits::{MsgSecretEntry, TcTokenEntry};
@@ -93,12 +93,6 @@ impl Client {
             return;
         }
 
-        // file_length is the decrypted (but still zlib-compressed) blob size, not
-        // the final decompressed size. We still pass it as a hint — the decompressor
-        // uses it with a 4x multiplier, which is a better estimate than guessing
-        // from the encrypted size (which includes MAC/padding overhead).
-        let compressed_size_hint = notification.file_length.filter(|&s| s > 0);
-
         // Use take() to avoid cloning large payloads - moves ownership instead
         let compressed_data = if let Some(inline_payload) =
             notification.initial_hist_bootstrap_inline_payload.take()
@@ -147,31 +141,23 @@ impl Client {
             device_snapshot.pn.as_ref().map(|j| j.to_non_ad().user)
         };
 
-        // Retain (and fully decompress) the blob only when a handler actually
-        // wants HistorySync. A message-only bot leaves this false, so the
-        // streaming decompress-and-parse path runs instead of materializing the
-        // whole payload just to drop it at dispatch.
-        let retain_history_blob = self.core.event_bus.has_handler_for(EventKind::HistorySync);
-
+        // Always carry the compressed input through (a move, no copy or extra
+        // inflate); handler interest is evaluated at dispatch time below, so a
+        // handler that registers while a large blob is being parsed still gets
+        // the event instead of racing a pre-parse snapshot.
         // 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(),
-                retain_history_blob,
-                compressed_size_hint,
+                true,
             ))
         } else {
             let (result_tx, result_rx) = futures::channel::oneshot::channel();
             let blocking_fut = self.runtime.spawn_blocking(Box::new(move || {
-                let result = process_history_sync(
-                    compressed_data,
-                    own_user.as_deref(),
-                    retain_history_blob,
-                    compressed_size_hint,
-                );
+                let result = process_history_sync(compressed_data, own_user.as_deref(), true);
                 let _ = result_tx.send(result);
             }));
             self.runtime
@@ -225,9 +211,15 @@ impl Client {
                 self.store_history_sync_msg_secrets(sync_result.msg_secret_records)
                     .await;
 
-                if let Some(decompressed) = sync_result.decompressed_bytes {
+                // No interest pre-check: dispatch() evaluates handler interest
+                // against a single bus snapshot (and skips materializing the
+                // Arc when nobody listens), so deferring to it removes the
+                // check-to-dispatch race window entirely. Building the event
+                // is just a Bytes refcount move plus metadata.
+                if let Some(compressed) = sync_result.compressed_bytes {
                     let lazy_hs = LazyHistorySync::new(
-                        decompressed,
+                        compressed,
+                        sync_result.decompressed_size,
                         notification.sync_type().into(),
                         notification.chunk_order,
                         notification.progress,
@@ -529,20 +521,20 @@ mod tests {
             conversations: vec![wa::Conversation {
                 id: chat.to_string(),
                 messages: vec![wa::HistorySyncMsg {
-                    message: Some(wa::WebMessageInfo {
+                    message: Some(Box::new(wa::WebMessageInfo {
                         key: wa::MessageKey {
                             remote_jid: Some(chat.to_string()),
                             from_me: Some(false),
                             id: Some(parent_id.to_string()),
                             participant: None,
                         },
-                        message: Some(wa::Message {
+                        message: Some(Box::new(wa::Message {
                             conversation: Some("historical".to_string()),
                             ..Default::default()
-                        }),
+                        })),
                         message_secret: Some(secret.clone()),
                         ..Default::default()
-                    }),
+                    })),
                     msg_order_id: Some(1),
                 }],
                 ..Default::default()
@@ -570,6 +562,57 @@ mod tests {
         assert_eq!(got, Some(secret));
     }
 
+    #[tokio::test]
+    async fn process_history_sync_task_dispatches_compressed_lazy_event() {
+        let client = crate::test_utils::create_test_client_with_name("history_lazy_event").await;
+        client.is_running.store(true, Ordering::Relaxed);
+
+        let chat = "5511777776666@s.whatsapp.net";
+        let history_sync = wa::HistorySync {
+            sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32,
+            conversations: vec![wa::Conversation {
+                id: chat.to_string(),
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        let raw_len = history_sync.encode_to_vec().len();
+        let compressed = compress_history_sync(&history_sync);
+        let compressed_copy = compressed.clone();
+        let notification = HistorySyncNotification {
+            file_length: Some(compressed.len() as u64),
+            sync_type: Some(wa::message::HistorySyncType::InitialBootstrap as i32),
+            initial_hist_bootstrap_inline_payload: Some(compressed),
+            ..Default::default()
+        };
+
+        // Register a handler BEFORE the task so retain_blob is true.
+        let (handler, event_rx) = wacore::types::events::ChannelEventHandler::new();
+        client.core.event_bus.add_handler(handler);
+
+        client
+            .process_history_sync_task("HIST_LAZY_EVENT".to_string(), notification)
+            .await;
+
+        let event = event_rx.try_recv().expect("HistorySync event dispatched");
+        let crate::types::events::Event::HistorySync(lazy) = &*event else {
+            panic!("expected HistorySync event, got {event:?}");
+        };
+
+        // The event carries the original compressed payload plus the exact
+        // inflated size, and every consumption path works.
+        assert_eq!(lazy.compressed_bytes().as_ref(), &compressed_copy[..]);
+        assert_eq!(lazy.decompressed_size(), raw_len);
+        let decoded = lazy.get().expect("decodes");
+        assert_eq!(decoded.conversations[0].id, chat);
+        let mut stream = lazy.stream();
+        assert_eq!(
+            stream.next_conversation().unwrap().unwrap().id,
+            chat,
+            "stream still works after get()"
+        );
+    }
+
     #[tokio::test]
     async fn process_history_sync_task_stores_bot_dm_secret_alias() {
         let client =
@@ -590,20 +633,20 @@ mod tests {
             conversations: vec![wa::Conversation {
                 id: chat.to_string(),
                 messages: vec![wa::HistorySyncMsg {
-                    message: Some(wa::WebMessageInfo {
+                    message: Some(Box::new(wa::WebMessageInfo {
                         key: wa::MessageKey {
                             remote_jid: Some(chat.to_string()),
                             from_me: Some(false),
                             id: Some(parent_id.to_string()),
                             participant: None,
                         },
-                        message: Some(wa::Message {
+                        message: Some(Box::new(wa::Message {
                             conversation: Some("bot historical".to_string()),
                             ..Default::default()
-                        }),
+                        })),
                         message_secret: Some(secret.clone()),
                         ..Default::default()
-                    }),
+                    })),
                     msg_order_id: Some(1),
                 }],
                 ..Default::default()
@@ -654,18 +697,18 @@ mod tests {
             }
         };
         wa::HistorySyncMsg {
-            message: Some(wa::WebMessageInfo {
+            message: Some(Box::new(wa::WebMessageInfo {
                 key: wa::MessageKey {
                     remote_jid: Some(chat.to_string()),
                     from_me: Some(false),
                     id: Some(msg_id.to_string()),
                     participant: None,
                 },
-                message: Some(message),
+                message: Some(Box::new(message)),
                 message_secret: Some(secret.to_vec()),
                 message_timestamp: Some(ts_secs),
                 ..Default::default()
-            }),
+            })),
             msg_order_id: Some(1),
         }
     }
@@ -936,33 +979,35 @@ mod tests {
         ts_secs: u64,
         bot_prompt: bool,
     ) -> wa::HistorySyncMsg {
-        let message_context_info = bot_prompt.then(|| wa::MessageContextInfo {
-            bot_metadata: Some(wa::BotMetadata {
-                persona_id: Some("867051314767696".into()),
+        let message_context_info = bot_prompt.then(|| {
+            Box::new(wa::MessageContextInfo {
+                bot_metadata: Some(wa::BotMetadata {
+                    persona_id: Some("867051314767696".into()),
+                    ..Default::default()
+                }),
                 ..Default::default()
-            }),
-            ..Default::default()
+            })
         });
         wa::HistorySyncMsg {
-            message: Some(wa::WebMessageInfo {
+            message: Some(Box::new(wa::WebMessageInfo {
                 key: wa::MessageKey {
                     remote_jid: Some(chat.to_string()),
                     from_me: Some(false),
                     id: Some(msg_id.to_string()),
                     participant: Some(participant.to_string()),
                 },
-                message: Some(wa::Message {
+                message: Some(Box::new(wa::Message {
                     extended_text_message: Some(Box::new(wa::message::ExtendedTextMessage {
                         text: Some("hi".into()),
                         ..Default::default()
                     })),
                     message_context_info,
                     ..Default::default()
-                }),
+                })),
                 message_secret: Some(secret.to_vec()),
                 message_timestamp: Some(ts_secs),
                 ..Default::default()
-            }),
+            })),
             msg_order_id: Some(1),
         }
     }