From 3ba44468dbd7cdb5e711da195eb6d3030f2544c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas=20de=20Oliveira=20Lopes?= Date: Thu, 22 Oct 2026 20:15:19 -0300 Subject: [PATCH] fix(chat-store): materialize deliberately shared group history --- .../down.sql | 2 + .../up.sql | 10 + crates/chat-store/src/error.rs | 3 + crates/chat-store/src/lib.rs | 1 + crates/chat-store/src/materialize.rs | 25 +- crates/chat-store/src/queries.rs | 6 + crates/chat-store/src/schema.rs | 1 + crates/chat-store/src/store/history_sync.rs | 27 +- crates/chat-store/src/store/inbound.rs | 29 +- .../chat-store/src/store/message_identity.rs | 13 +- crates/chat-store/src/store/message_rows.rs | 31 + crates/chat-store/src/store/mod.rs | 26 +- crates/chat-store/src/store/read_state.rs | 6 + crates/chat-store/src/store/shared_history.rs | 439 ++++++++++++++ crates/chat-store/src/store/writer.rs | 6 + crates/chat-store/src/types.rs | 6 + crates/chat-store/tests/regression_245.rs | 119 ++++ crates/chat-store/tests/shared_history.rs | 535 ++++++++++++++++++ crates/session/src/whatsapp/convert.rs | 31 +- crates/session/src/whatsapp/mod.rs | 5 +- crates/session/src/whatsapp/shared_history.rs | 127 +++++ crates/session/src/whatsapp/tests.rs | 2 + docs/shared-group-history.md | 146 +++++ 23 files changed, 1582 insertions(+), 14 deletions(-) create mode 100644 crates/chat-store/migrations/2026-10-22-100000_shared_group_history/down.sql create mode 100644 crates/chat-store/migrations/2026-10-22-100000_shared_group_history/up.sql create mode 100644 crates/chat-store/src/store/shared_history.rs create mode 100644 crates/chat-store/tests/regression_245.rs create mode 100644 crates/chat-store/tests/shared_history.rs create mode 100644 crates/session/src/whatsapp/shared_history.rs create mode 100644 docs/shared-group-history.md diff --git a/crates/chat-store/migrations/2026-10-22-100000_shared_group_history/down.sql b/crates/chat-store/migrations/2026-10-22-100000_shared_group_history/down.sql new file mode 100644 index 00000000..54cd41ee --- /dev/null +++ b/crates/chat-store/migrations/2026-10-22-100000_shared_group_history/down.sql @@ -0,0 +1,2 @@ +DROP TABLE shared_group_history; +ALTER TABLE messages DROP COLUMN received_as_shared_history; diff --git a/crates/chat-store/migrations/2026-10-22-100000_shared_group_history/up.sql b/crates/chat-store/migrations/2026-10-22-100000_shared_group_history/up.sql new file mode 100644 index 00000000..aa78e3e8 --- /dev/null +++ b/crates/chat-store/migrations/2026-10-22-100000_shared_group_history/up.sql @@ -0,0 +1,10 @@ +CREATE TABLE shared_group_history ( + device_id INTEGER NOT NULL REFERENCES device(id) ON DELETE CASCADE, + chat_jid TEXT NOT NULL, msg_id TEXT NOT NULL, sender_jid TEXT NOT NULL, + timestamp_ms BIGINT NOT NULL, proto BLOB NOT NULL, + attempted_ms BIGINT NOT NULL DEFAULT 0, + completed BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY(device_id, chat_jid, msg_id, sender_jid) +); +ALTER TABLE messages ADD COLUMN received_as_shared_history BOOLEAN NOT NULL DEFAULT FALSE; +CREATE INDEX shared_group_history_pending ON shared_group_history(device_id, completed, attempted_ms, timestamp_ms, msg_id); diff --git a/crates/chat-store/src/error.rs b/crates/chat-store/src/error.rs index a85c2073..46ee637e 100644 --- a/crates/chat-store/src/error.rs +++ b/crates/chat-store/src/error.rs @@ -7,6 +7,9 @@ pub enum ChatStoreError { #[error("storage error")] Store(#[from] StoreError), + #[error("invalid shared group history: {0}")] + InvalidGroupHistory(String), + #[error("invalid full-text search query")] InvalidSearchQuery, diff --git a/crates/chat-store/src/lib.rs b/crates/chat-store/src/lib.rs index 0c5b3ebe..0c20663b 100644 --- a/crates/chat-store/src/lib.rs +++ b/crates/chat-store/src/lib.rs @@ -53,6 +53,7 @@ pub use materialize::{ }; pub use queries::MessageReadState; pub use store::ChatStore; +pub use store::shared_history::PendingHistoryBundle; pub use types::{ ArrivalCursor, AvatarDescriptor, ChatArchiveFilter, ChatCursor, ChatEntry, ChatNameExpected, ChatNameWrite, ChatNotificationMetadata, ContactEntry, GroupHierarchyWrite, MediaRef, diff --git a/crates/chat-store/src/materialize.rs b/crates/chat-store/src/materialize.rs index e53878bb..695f7122 100644 --- a/crates/chat-store/src/materialize.rs +++ b/crates/chat-store/src/materialize.rs @@ -34,6 +34,8 @@ pub(crate) enum MessageOp { target_from_me: bool, target_participant: Option, }, + /// A deliberately shared group-history attachment, persisted before ACK. + HistoryBundle, /// Protocol/bookkeeping payloads that don't belong in a chat log. Ignore, } @@ -68,6 +70,8 @@ pub fn message_kind(message: &wa::Message) -> &'static str { "location" } else if poll_creation_message(message).is_some() { "poll" + } else if base.message_history_notice.is_set() { + "history_notice" } else if base.album_message.is_set() { "album" } else if base.product_message.is_set() { @@ -242,6 +246,10 @@ pub(crate) fn classify(msg: &wa::Message) -> MessageOp { } } + if base.message_history_bundle.is_set() { + return MessageOp::HistoryBundle; + } + if is_control_carrier(base) { return MessageOp::Ignore; } @@ -260,6 +268,20 @@ pub(crate) fn classify(msg: &wa::Message) -> MessageOp { /// Text projection for list previews and full-text search: body text or media /// caption. pub(crate) fn extract_text(base: &wa::Message) -> Option { + if let Some(notice) = base.message_history_notice.as_option() { + let metadata = notice.message_history_metadata.as_option(); + return Some( + match metadata.and_then(|m| m.message_count).filter(|n| *n >= 0) { + Some(count) => { + let receivers = metadata.map_or(0, |m| m.history_receivers.len()); + let messages = if count == 1 { "message" } else { "messages" }; + let members = if receivers == 1 { "member" } else { "members" }; + format!("Shared {count} earlier {messages} with {receivers} group {members}") + } + None => "Shared earlier messages with group members".into(), + }, + ); + } base.text_content() .or_else(|| base.get_caption()) .or_else(|| business_text(base)) @@ -337,7 +359,8 @@ fn has_any_content(base: &wa::Message) -> bool { /// a carrier. #[must_use] pub fn is_control_only(msg: &wa::Message) -> bool { - is_control_carrier(normalized_message(msg)) + let base = normalized_message(msg); + is_control_carrier(base) || base.message_history_bundle.is_set() } #[cfg(test)] diff --git a/crates/chat-store/src/queries.rs b/crates/chat-store/src/queries.rs index 5b6ca20a..fc31961e 100644 --- a/crates/chat-store/src/queries.rs +++ b/crates/chat-store/src/queries.rs @@ -243,6 +243,7 @@ pub(crate) struct MessageRow { edited_at_ms: Option, revoked: bool, local_revoke_placeholder: bool, + received_as_shared_history: bool, } /// Fold an author-equivalent legacy copy for a read without hiding a @@ -298,6 +299,8 @@ fn fold_read_duplicate( held.status }; survivor.revoked = revoked; + survivor.received_as_shared_history = + held.received_as_shared_history && incoming.received_as_shared_history; survivor.local_revoke_placeholder = held.local_revoke_placeholder || incoming.local_revoke_placeholder; survivor.status = status; @@ -492,6 +495,7 @@ impl From for StoredMessage { edited_at: row.edited_at_ms.and_then(ms_to_utc), revoked: row.revoked, local_revoke_placeholder: row.local_revoke_placeholder, + received_as_shared_history: row.received_as_shared_history, seq: row.id, } } @@ -937,6 +941,8 @@ impl ChatStore { .and(dsl::chat_jid.eq_any(keys)) .and(dsl::from_me.eq(false)) .and(dsl::local_revoke_placeholder.eq(false)) + .and(dsl::received_as_shared_history.eq(false)) + .and(dsl::kind.ne("history_notice")) .and( dsl::timestamp_ms .gt(cursor.timestamp_ms) diff --git a/crates/chat-store/src/schema.rs b/crates/chat-store/src/schema.rs index d0c7b217..ebab96d0 100644 --- a/crates/chat-store/src/schema.rs +++ b/crates/chat-store/src/schema.rs @@ -46,6 +46,7 @@ diesel::table! { edited_at_ms -> Nullable, revoked -> Bool, local_revoke_placeholder -> Bool, + received_as_shared_history -> Bool, } } diff --git a/crates/chat-store/src/store/history_sync.rs b/crates/chat-store/src/store/history_sync.rs index 6ab50168..9971c43b 100644 --- a/crates/chat-store/src/store/history_sync.rs +++ b/crates/chat-store/src/store/history_sync.rs @@ -172,7 +172,7 @@ fn apply_history_conversation( let Some(wmi) = hist_msg.message.as_option() else { continue; }; - apply_history_message(conn, device_id, chat, wmi, cs, &mut pending_quotes)?; + apply_history_message(conn, device_id, chat, wmi, cs, &mut pending_quotes, false)?; } strip_pending_quotes(conn, device_id, chat, &pending_quotes)?; // Backfill the denormalized preview from the newest materialized row, so a @@ -311,13 +311,14 @@ fn apply_history_conversation( Ok(()) } -fn apply_history_message( +pub(super) fn apply_history_message( conn: &mut SqliteConnection, device_id: i32, chat: &str, wmi: &wa::WebMessageInfo, cs: &mut ChangeSet, pending_quotes: &mut Vec, + shared: bool, ) -> QueryResult<()> { let Some(key) = wmi.key.as_option() else { return Ok(()); @@ -348,6 +349,11 @@ fn apply_history_message( if let Some(message) = wmi.message.as_option() { match classify(message) { MessageOp::Store { kind, text } => { + if kind == "history_notice" + && !super::shared_history::notice_visible(conn, device_id, message, from_me)? + { + return Ok(()); + } // Same compaction as the live path. A parent later in this // same conversation is not visible yet, so the post-pass // below re-checks the replies that kept their snapshot. @@ -381,6 +387,21 @@ fn apply_history_message( }, cs, )?; + if shared && stored_row == StoredRow::Inserted { + use schema::messages::dsl; + diesel::update( + dsl::messages + .filter(dsl::device_id.eq(device_id)) + .filter(dsl::chat_jid.eq(chat)) + .filter(dsl::msg_id.eq(msg_id)) + .filter(dsl::sender_jid.eq( + crate::store::message_identity::stored_sender(sender, from_me), + )) + .filter(dsl::from_me.eq(from_me)), + ) + .set(dsl::received_as_shared_history.eq(true)) + .execute(conn)?; + } // A reply whose parent lands LATER in this conversation kept // its inline snapshot above (the parent was not visible yet). // Record it for the post-pass rather than re-reading every @@ -440,7 +461,7 @@ fn apply_history_message( cs.chats = true; } } - MessageOp::Ignore => {} + MessageOp::HistoryBundle | MessageOp::Ignore => {} } } diff --git a/crates/chat-store/src/store/inbound.rs b/crates/chat-store/src/store/inbound.rs index 7a7d03db..c142ea14 100644 --- a/crates/chat-store/src/store/inbound.rs +++ b/crates/chat-store/src/store/inbound.rs @@ -47,6 +47,16 @@ pub(super) fn apply_inbound( match classify(&inbound.message) { MessageOp::Store { kind, text } => { + if kind == "history_notice" + && !super::shared_history::notice_visible( + conn, + device_id, + &inbound.message, + info.source.is_from_me, + )? + { + return Ok(()); + } // Compacted storage copy (quote snapshot dropped when the parent // is local, secret-only envelope stripped, large protos // compressed); the live message above is untouched. @@ -81,8 +91,11 @@ pub(super) fn apply_inbound( // already counted) must not inflate the unread badge again — and a // skipped one (revoked tombstone) must not surface its content in // the chat preview at all. - let unread_delta = - i32::from(inserted == StoredRow::Inserted && !info.source.is_from_me); + let unread_delta = i32::from( + inserted == StoredRow::Inserted + && !info.source.is_from_me + && kind != "history_notice", + ); if inserted != StoredRow::Skipped { bump_chat( conn, @@ -173,6 +186,18 @@ pub(super) fn apply_inbound( } cs.message_chats.insert(chat); } + MessageOp::HistoryBundle => { + super::shared_history::persist( + conn, + device_id, + &chat, + &info.id, + &sender, + ts_ms, + info.source.is_from_me, + &inbound.message, + )?; + } MessageOp::Ignore => {} } Ok(()) diff --git a/crates/chat-store/src/store/message_identity.rs b/crates/chat-store/src/store/message_identity.rs index 96ab4059..d58cb1a1 100644 --- a/crates/chat-store/src/store/message_identity.rs +++ b/crates/chat-store/src/store/message_identity.rs @@ -134,6 +134,8 @@ struct StoredMessage { revoked: bool, #[diesel(sql_type = Bool)] local_revoke_placeholder: bool, + #[diesel(sql_type = Bool)] + received_as_shared_history: bool, } /// Fold rows already proven to be one author/message. Keep the oldest stable @@ -169,6 +171,7 @@ pub(crate) fn merge_rows( dsl::edited_at_ms, dsl::revoked, dsl::local_revoke_placeholder, + dsl::received_as_shared_history, )) .load(conn)?; rows.sort_by_key(|row| row.id); @@ -253,6 +256,8 @@ pub(crate) fn merge_rows( dsl::edited_at_ms.eq(edited_at_ms), dsl::revoked.eq(tombstone), dsl::local_revoke_placeholder.eq(rows.iter().any(|row| row.local_revoke_placeholder)), + dsl::received_as_shared_history + .eq(rows.iter().all(|row| row.received_as_shared_history)), )) .execute(conn)?; Ok(MessageOwner { @@ -403,7 +408,7 @@ fn backfill_legacy_read_state(conn: &mut SqliteConnection, device_id: i32) -> Qu AND c.read_boundary_ms = 0 AND c.read_boundary_ids IS NULL AND m.from_me = FALSE - AND m.local_revoke_placeholder = FALSE + AND m.local_revoke_placeholder = FALSE AND m.received_as_shared_history = FALSE AND m.kind != 'history_notice' AND m.timestamp_ms > ?", ) .bind::(device_id) @@ -424,7 +429,7 @@ fn backfill_legacy_read_state(conn: &mut SqliteConnection, device_id: i32) -> Qu WHERE messages.device_id = chats.device_id AND messages.chat_jid = chats.jid AND messages.from_me = FALSE - AND messages.local_revoke_placeholder = FALSE + AND messages.local_revoke_placeholder = FALSE AND messages.received_as_shared_history = FALSE AND messages.kind != 'history_notice' ) WHERE chats.device_id = ? AND chats.unread_count = 0 @@ -435,14 +440,14 @@ fn backfill_legacy_read_state(conn: &mut SqliteConnection, device_id: i32) -> Qu WHERE messages.device_id = chats.device_id AND messages.chat_jid = chats.jid AND messages.from_me = FALSE - AND messages.local_revoke_placeholder = FALSE + AND messages.local_revoke_placeholder = FALSE AND messages.received_as_shared_history = FALSE AND messages.kind != 'history_notice' ) AND ( SELECT COUNT(DISTINCT msg_id) FROM messages WHERE messages.device_id = chats.device_id AND messages.chat_jid = chats.jid AND messages.from_me = FALSE - AND messages.local_revoke_placeholder = FALSE + AND messages.local_revoke_placeholder = FALSE AND messages.received_as_shared_history = FALSE AND messages.kind != 'history_notice' AND messages.timestamp_ms > ? ) <= ?", ) diff --git a/crates/chat-store/src/store/message_rows.rs b/crates/chat-store/src/store/message_rows.rs index 0cd55037..a14a7c92 100644 --- a/crates/chat-store/src/store/message_rows.rs +++ b/crates/chat-store/src/store/message_rows.rs @@ -61,6 +61,37 @@ pub(super) fn insert_message( &sender, changes, )? { + // A revoke that arrived first had no original position. Recover only + // its kind/time, never the content, after sender-scoped resolution. + let old_ts: i64 = dsl::messages + .filter(dsl::id.eq(existing.id)) + .select(dsl::timestamp_ms) + .first(conn)?; + let repaired = diesel::update( + dsl::messages + .filter(dsl::id.eq(existing.id)) + .filter(dsl::revoked.eq(true)) + .filter(dsl::kind.eq("unknown")) + .filter(dsl::proto.is_null()), + ) + .set(( + dsl::timestamp_ms.eq(new.timestamp_ms), + dsl::kind.eq(new.kind), + )) + .execute(conn)?; + if repaired > 0 { + // Re-derive the activity head too when this placeholder was newest. + super::chat_rows::reconcile_chat_head_after_timestamp_change( + conn, + device_id, + new.chat_jid, + old_ts, + new.timestamp_ms, + )?; + changes.chats = true; + changes.message_chats.insert(new.chat_jid.to_owned()); + return Ok(StoredRow::Skipped); + } if !new.overwrite { // History is a stale copy: live rows and placeholders win. return Ok(StoredRow::Skipped); diff --git a/crates/chat-store/src/store/mod.rs b/crates/chat-store/src/store/mod.rs index 151602f2..fcc7bcd3 100644 --- a/crates/chat-store/src/store/mod.rs +++ b/crates/chat-store/src/store/mod.rs @@ -23,6 +23,7 @@ mod reaction; pub(crate) mod read_state; mod receipt; mod revoke; +pub(crate) mod shared_history; mod writer; use std::sync::{Arc, Mutex}; @@ -160,6 +161,11 @@ pub(crate) enum WriterMsg { emoji: String, timestamp_ms: i64, }, + SharedHistoryAttempt(crate::PendingHistoryBundle), + SharedHistory { + bundle: crate::PendingHistoryBundle, + history: Option, + }, Reconcile(Jid), ReconcileAll, ReconcileMappings(Vec<(String, String)>), @@ -450,7 +456,7 @@ struct MetadataValue { fn repair_unknown_message_kinds(conn: &mut diesel::SqliteConnection) -> QueryResult<()> { use diesel::sql_types::{BigInt, Nullable, Text}; const KEY: &str = "message_kind_classifier"; - const VERSION: &str = "2026-09-24-v1"; + const VERSION: &str = "2026-10-22-v2"; conn.transaction(|conn| { let version = diesel::sql_query("SELECT value FROM chat_store_meta WHERE key = ?") .bind::(KEY) @@ -480,6 +486,16 @@ fn repair_unknown_message_kinds(conn: &mut diesel::SqliteConnection) -> QueryRes else { continue; }; + if shared_history::repair_legacy( + conn, + row.device_id, + row.id, + &row.chat_jid, + &message, + )? { + affected_chats.insert((row.device_id, row.chat_jid)); + continue; + } let kind = crate::materialize::message_kind(&message); if kind == "unknown" { continue; @@ -1191,6 +1207,14 @@ mod migration_tests { .expect("create store"); ChatStore::new(&store).await.expect("run migrations"); + assert!(has_table(&store, "shared_group_history").await); + store + .shared() + .run(revert_last_chat_migration) + .await + .expect("shared-history downgrade is reversible"); + assert!(!has_table(&store, "shared_group_history").await); + assert!(!has_column(&store, "messages", "received_as_shared_history").await); assert!(has_column(&store, "messages", "local_revoke_placeholder").await); store .shared() diff --git a/crates/chat-store/src/store/read_state.rs b/crates/chat-store/src/store/read_state.rs index 1df55c11..7a4c20c2 100644 --- a/crates/chat-store/src/store/read_state.rs +++ b/crates/chat-store/src/store/read_state.rs @@ -190,6 +190,8 @@ pub(super) fn advance_read_state( .and(dsl::msg_id.eq_any(&state.extra_ids)) .and(dsl::from_me.eq(false)) .and(dsl::local_revoke_placeholder.eq(false)) + .and(dsl::received_as_shared_history.eq(false)) + .and(dsl::kind.ne("history_notice")) .and(dsl::timestamp_ms.gt(state.watermark_ms)), ) .select(dsl::msg_id) @@ -261,6 +263,8 @@ pub(super) fn count_unread( .and(dsl::chat_jid.eq(chat)) .and(dsl::from_me.eq(false)) .and(dsl::local_revoke_placeholder.eq(false)) + .and(dsl::received_as_shared_history.eq(false)) + .and(dsl::kind.ne("history_notice")) .and(dsl::timestamp_ms.gt(state.watermark_ms)), ) .into_boxed(); @@ -290,6 +294,8 @@ pub(super) fn count_uncovered_incoming( .and(dsl::chat_jid.eq(chat)) .and(dsl::from_me.eq(false)) .and(dsl::local_revoke_placeholder.eq(false)) + .and(dsl::received_as_shared_history.eq(false)) + .and(dsl::kind.ne("history_notice")) .and(dsl::timestamp_ms.gt(state.watermark_ms)), ) .into_boxed(); diff --git a/crates/chat-store/src/store/shared_history.rs b/crates/chat-store/src/store/shared_history.rs new file mode 100644 index 00000000..91fac4dc --- /dev/null +++ b/crates/chat-store/src/store/shared_history.rs @@ -0,0 +1,439 @@ +//! Durable, account-scoped work for history explicitly shared with this companion. + +use diesel::{ + prelude::*, + sql_types::{BigInt, Binary, Integer, Text}, +}; +use wacore::proto_helpers::MessageExt; +use wacore_binary::{Jid, JidExt as _}; +use waproto::whatsapp as wa; + +use super::{ + ChatStore, WriterMsg, + message_identity::{own_participant_jids, stored_sender}, + writer::ChangeSet, +}; +use crate::{ChatStoreError, Result, db_err, normalized_message}; + +/// A persisted attachment. Keep this value intact when completing or discarding it. +#[derive(Clone, QueryableByName)] +pub struct PendingHistoryBundle { + #[diesel(sql_type = Text)] + pub chat_jid: String, + #[diesel(sql_type = Text)] + pub msg_id: String, + #[diesel(sql_type = Text)] + pub sender_jid: String, + #[diesel(sql_type = BigInt)] + pub timestamp_ms: i64, + #[diesel(sql_type = Binary)] + proto: Vec, +} +impl std::fmt::Debug for PendingHistoryBundle { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PendingHistoryBundle") + .finish_non_exhaustive() + } +} +impl PendingHistoryBundle { + pub fn message(&self) -> Result { + waproto::codec::message_decode(&self.proto) + .map_err(|e| ChatStoreError::InvalidGroupHistory(e.to_string())) + } +} + +// Local resource ceilings, not claims about server limits. +const MAX_BYTES: u64 = 32 * 1024 * 1024; +const MAX_MESSAGES: usize = 10_000; + +impl ChatStore { + /// Bounded page of unconsumed attachments, including work recovered after restart. + pub async fn pending_history_bundles(&self) -> Result> { + let device = self.device_id; + Ok(self.db.run(move |conn| { + diesel::sql_query("SELECT chat_jid, msg_id, sender_jid, timestamp_ms, proto FROM shared_group_history WHERE device_id = ? AND completed = FALSE ORDER BY attempted_ms, timestamp_ms, msg_id LIMIT 32") + .bind::(device).load(conn).map_err(db_err) + }).await?) + } + + /// Queue the authenticated download's zlib/protobuf body. Consumption and + /// stale-history insertion commit together. Call `flush` to observe durability. + pub fn apply_history_bundle( + &self, + bundle: &PendingHistoryBundle, + compressed: &[u8], + ) -> Result<()> { + let invalid = ChatStoreError::InvalidGroupHistory; + if compressed.len() as u64 > MAX_BYTES { + return Err(invalid("compressed size limit".into())); + } + let bytes = inflate(compressed).map_err(invalid)?; + preflight(&bytes).map_err(invalid)?; + let history = + waproto::codec::group_history_decode(&bytes).map_err(|e| invalid(e.to_string()))?; + if history.messages.len() > MAX_MESSAGES { + return Err(invalid("message count limit".into())); + } + self.tx + .send(WriterMsg::SharedHistory { + bundle: bundle.clone(), + history: Some(history), + }) + .map_err(|e| ChatStoreError::WriteBatchFailed(e.into())) + } + + /// Rotate a failed attachment behind untouched work without losing it. + pub fn record_history_bundle_attempt(&self, bundle: &PendingHistoryBundle) -> Result<()> { + self.tx + .send(WriterMsg::SharedHistoryAttempt(bundle.clone())) + .map_err(|e| ChatStoreError::WriteBatchFailed(e.into())) + } + + /// Retire an unusable attachment without inserting a timeline placeholder. + /// A transient download failure should leave the attachment pending instead. + pub fn discard_history_bundle(&self, bundle: &PendingHistoryBundle) -> Result<()> { + self.tx + .send(WriterMsg::SharedHistory { + bundle: bundle.clone(), + history: None, + }) + .map_err(|e| ChatStoreError::WriteBatchFailed(e.into())) + } +} + +#[allow(clippy::too_many_arguments)] +pub(super) fn persist( + conn: &mut SqliteConnection, + device: i32, + chat: &str, + id: &str, + sender: &str, + ts: i64, + from_me: bool, + message: &wa::Message, +) -> QueryResult<()> { + if from_me || !chat.parse::().is_ok_and(|j| j.is_group()) { + return Ok(()); + } + let Some(bundle) = normalized_message(message) + .message_history_bundle + .as_option() + else { + return Ok(()); + }; + let Some(metadata) = bundle.message_history_metadata.as_option() else { + return Ok(()); + }; + if id.is_empty() + || !sender.parse::().is_ok_and(|j| j.is_pn() || j.is_lid()) + || bundle.mimetype.as_deref() != Some("application/protobuf") + || bundle.direct_path.as_deref().is_none_or(str::is_empty) + || bundle.media_key.as_ref().is_none_or(|b| b.len() != 32) + || bundle.file_sha256.as_ref().is_none_or(|b| b.len() != 32) + || bundle + .file_enc_sha256 + .as_ref() + .is_none_or(|b| b.len() != 32) + || metadata.oldest_message_timestamp_in_window.is_none() + || metadata.message_count.is_none_or(|n| n <= 0) + { + return Ok(()); + } + let own = own_participant_jids(conn, device)?; + if !metadata + .history_receivers + .iter() + .any(|j| own.contains(&stored_sender(j, false))) + { + return Ok(()); + } + let sender = stored_sender(sender, false); + diesel::sql_query("INSERT OR IGNORE INTO shared_group_history (device_id, chat_jid, msg_id, sender_jid, timestamp_ms, proto) VALUES (?, ?, ?, ?, ?, ?)") + .bind::(device).bind::(chat).bind::(id).bind::(sender).bind::(ts) + .bind::(waproto::codec::message_to_vec(message)).execute(conn)?; + Ok(()) +} + +pub(super) fn apply( + conn: &mut SqliteConnection, + device: i32, + bundle: &PendingHistoryBundle, + history: Option<&wa::GroupHistory>, + cs: &mut ChangeSet, +) -> QueryResult<()> { + // Compare the complete envelope too: a stale completion cannot consume a + // different attachment, even if a sender reused its stanza id. + let changed = diesel::sql_query("UPDATE shared_group_history SET completed = TRUE, proto = X'' WHERE device_id = ? AND chat_jid = ? AND msg_id = ? AND sender_jid = ? AND proto = ? AND timestamp_ms = ? AND completed = FALSE") + .bind::(device).bind::(&bundle.chat_jid).bind::(&bundle.msg_id).bind::(&bundle.sender_jid).bind::(&bundle.proto).bind::(bundle.timestamp_ms).execute(conn)?; + if changed == 0 { + return Ok(()); + } + let Some(history) = history else { + return Ok(()); + }; + let envelope = bundle + .message() + .map_err(|e| diesel::result::Error::DeserializationError(Box::new(e)))?; + let Some(metadata) = normalized_message(&envelope) + .message_history_bundle + .as_option() + .and_then(|b| b.message_history_metadata.as_option()) + else { + return Ok(()); + }; + let own = own_participant_jids(conn, device)?; + let mut quotes = Vec::new(); + for original in &history.messages { + let mut message = original.clone(); + let Some(key) = message.key.as_option_mut() else { + continue; + }; + if key.remote_jid.as_deref() != Some(&bundle.chat_jid) + || key.id.as_deref().is_none_or(str::is_empty) + { + continue; + } + let author = if key.from_me == Some(true) { + Some(bundle.sender_jid.as_str()) + } else { + message + .participant + .as_deref() + .or(key.participant.as_deref()) + .or_else(|| own.first().map(String::as_str)) + }; + let Some(author) = + author.filter(|s| s.parse::().is_ok_and(|j| j.is_pn() || j.is_lid())) + else { + continue; + }; + let author = stored_sender(author, false); + key.from_me = Some(own.contains(&author)); + key.participant = Some(author.clone()); + message.participant = Some(author); + if key.from_me == Some(true) && message.status.is_none() { + message.status = Some(wa::web_message_info::Status::SERVER_ACK); + } + let Some(ts) = message + .message_timestamp + .and_then(|t| i64::try_from(t).ok()) + .filter(|t| *t > 0) + else { + continue; + }; + if metadata + .oldest_message_timestamp_in_window + .is_some_and(|old| ts < old) + || ts > bundle.timestamp_ms / 1000 + || metadata + .oldest_message_timestamp_in_bundle + .is_some_and(|old| ts < old) + { + continue; + } + let Some(content) = message.message.as_option_mut() else { + continue; + }; + let base = normalized_message(content); + // Never recursively download or inject restricted view-once content. + if base.message_history_bundle.is_set() + || base.message_history_notice.is_set() + || content.is_view_once() + || base.is_view_once() + { + continue; + } + if let Some(context) = base.context_info() + && context.expiration.is_some_and(|duration| { + duration > 0 + && ts.saturating_add(i64::from(duration)) <= wacore::time::now_utc().timestamp() + }) + { + continue; + } + *content = normalized_message(content).clone(); + // Keys in a shared bundle use the sharing account's perspective. + if let Some(protocol) = content.protocol_message.as_option_mut() + && let Some(target) = protocol.key.as_option_mut() + { + let author = if target.from_me == Some(true) { + Some(bundle.sender_jid.as_str()) + } else { + target.participant.as_deref() + }; + let Some(author) = author else { + continue; + }; + let author = stored_sender(author, false); + target.from_me = Some(own.contains(&author)); + target.participant = Some(author); + } + if crate::message_kind(content) == "unknown" && content.protocol_message.is_unset() { + continue; + } + super::history_sync::apply_history_message( + conn, + device, + &bundle.chat_jid, + &message, + cs, + &mut quotes, + true, + )?; + } + crate::storage_proto::strip_pending_quotes(conn, device, &bundle.chat_jid, "es)?; + // History fills gaps without adding unread attention or overwriting a live head. + super::chat_rows::bump_chat( + conn, + device, + &bundle.chat_jid, + super::chat_rows::ChatBump { + msg_id: "", + ts_ms: 0, + preview: None, + kind: None, + unread_delta: 0, + }, + )?; + diesel::sql_query("UPDATE chats SET last_message_ts = MAX(last_message_ts, COALESCE((SELECT MAX(timestamp_ms) FROM messages WHERE device_id = ? AND chat_jid = ?), 0)) WHERE device_id = ? AND jid = ?") + .bind::(device).bind::(&bundle.chat_jid).bind::(device).bind::(&bundle.chat_jid).execute(conn)?; + super::chat_rows::recompute_chat_preview(conn, device, &bundle.chat_jid)?; + cs.chats = true; + cs.message_chats.insert(bundle.chat_jid.clone()); + Ok(()) +} + +pub(super) fn attempted( + conn: &mut SqliteConnection, + device: i32, + bundle: &PendingHistoryBundle, +) -> QueryResult<()> { + diesel::sql_query("UPDATE shared_group_history SET attempted_ms = ? WHERE device_id = ? AND chat_jid = ? AND msg_id = ? AND sender_jid = ?") + .bind::(wacore::time::now_utc().timestamp_millis()) + .bind::(device).bind::(&bundle.chat_jid).bind::(&bundle.msg_id).bind::(&bundle.sender_jid).execute(conn)?; + Ok(()) +} + +fn inflate(compressed: &[u8]) -> std::result::Result, String> { + let mut decoder = flate2::Decompress::new(true); + let mut bytes = Vec::new(); + loop { + let mut chunk = [0; 8192]; + let before_in = decoder.total_in(); + let before_out = decoder.total_out(); + let status = decoder + .decompress( + &compressed[before_in as usize..], + &mut chunk, + flate2::FlushDecompress::Finish, + ) + .map_err(|e| e.to_string())?; + let written = (decoder.total_out() - before_out) as usize; + if decoder.total_out() > MAX_BYTES { + return Err("inflated size limit".into()); + } + bytes.extend_from_slice(&chunk[..written]); + if status == flate2::Status::StreamEnd { + return if decoder.total_in() as usize == compressed.len() { + Ok(bytes) + } else { + Err("trailing zlib bytes".into()) + }; + } + if decoder.total_in() == before_in && written == 0 { + return Err("incomplete zlib stream".into()); + } + } +} + +pub(super) fn notice_visible( + conn: &mut SqliteConnection, + device: i32, + message: &wa::Message, + from_me: bool, +) -> QueryResult { + if from_me { + return Ok(false); + } + let Some(metadata) = normalized_message(message) + .message_history_notice + .as_option() + .and_then(|n| n.message_history_metadata.as_option()) + else { + return Ok(false); + }; + let own = own_participant_jids(conn, device)?; + Ok(metadata.message_count.is_some_and(|n| n > 0) + && metadata.oldest_message_timestamp_in_window.is_some() + && !metadata + .history_receivers + .iter() + .any(|j| own.contains(&stored_sender(j, false)))) +} + +// Reject a flood of empty top-level records before allocating WebMessageInfo +// objects. The generated codec still owns protobuf validation and decoding. +fn preflight(mut bytes: &[u8]) -> std::result::Result<(), String> { + let mut fields = 0; + while !bytes.is_empty() { + fields += 1; + if fields > MAX_MESSAGES { + return Err("record count limit".into()); + } + let tag = buffa::encoding::Tag::decode(&mut bytes).map_err(|e| e.to_string())?; + buffa::encoding::skip_field(tag, &mut bytes).map_err(|e| e.to_string())?; + } + Ok(()) +} + +/// Upgrade only classifier candidates; the caller owns the repair transaction +/// and refreshes each affected chat's preview after the page is repaired. +pub(super) fn repair_legacy( + conn: &mut SqliteConnection, + device: i32, + row_id: i64, + chat: &str, + message: &wa::Message, +) -> QueryResult { + use crate::schema::messages::dsl; + let base = normalized_message(message); + if base.message_history_bundle.is_unset() && base.message_history_notice.is_unset() { + return Ok(false); + } + let (id, sender, ts, own): (String, String, i64, bool) = dsl::messages + .filter(dsl::id.eq(row_id)) + .select(( + dsl::msg_id, + dsl::sender_jid, + dsl::timestamp_ms, + dsl::from_me, + )) + .first(conn)?; + if base.message_history_notice.is_set() && notice_visible(conn, device, message, own)? { + diesel::update(dsl::messages.filter(dsl::id.eq(row_id))) + .set(( + dsl::kind.eq("history_notice"), + dsl::text_content.eq(crate::materialize::extract_text(base)), + )) + .execute(conn)?; + } else { + persist(conn, device, chat, &id, &sender, ts, own, message)?; + diesel::delete(dsl::messages.filter(dsl::id.eq(row_id))).execute(conn)?; + } + // Remove only the carrier's unread contribution, respecting a read cursor + // and the negative manual-unread marker. + let state = super::read_state::read_state(conn, device, chat)?; + if !own && !state.covers(ts, &id) { + diesel::sql_query("UPDATE chats SET unread_count = MAX(0, unread_count - 1) WHERE device_id = ? AND jid = ? AND unread_count > 0") + .bind::(device).bind::(chat).execute(conn)?; + } + let head: Option = dsl::messages + .filter(dsl::device_id.eq(device)) + .filter(dsl::chat_jid.eq(chat)) + .select(diesel::dsl::max(dsl::timestamp_ms)) + .first(conn)?; + diesel::sql_query("UPDATE chats SET last_message_ts = ? WHERE device_id = ? AND jid = ? AND last_message_ts = ?") + .bind::(head.unwrap_or(0)).bind::(device) + .bind::(chat).bind::(ts).execute(conn)?; + Ok(true) +} diff --git a/crates/chat-store/src/store/writer.rs b/crates/chat-store/src/store/writer.rs index 2584ed0a..d1d409b9 100644 --- a/crates/chat-store/src/store/writer.rs +++ b/crates/chat-store/src/store/writer.rs @@ -393,6 +393,12 @@ fn apply_writer_msg( } Ok(()) } + WriterMsg::SharedHistoryAttempt(bundle) => { + super::shared_history::attempted(conn, device_id, bundle) + } + WriterMsg::SharedHistory { bundle, history } => { + super::shared_history::apply(conn, device_id, bundle, history.as_ref(), cs) + } WriterMsg::ReconcileAll => { crate::store::message_identity::reconcile_all(conn, device_id, cs)?; crate::lid::reconcile_known_chats(conn, device_id, cs)?; diff --git a/crates/chat-store/src/types.rs b/crates/chat-store/src/types.rs index 34e35b16..38bc6133 100644 --- a/crates/chat-store/src/types.rs +++ b/crates/chat-store/src/types.rs @@ -24,6 +24,8 @@ pub enum MessageKind { Poll, /// Parent for a grouped album; child media rows carry their own kinds. Album, + /// A group system notice describing deliberately shared history. + HistoryNotice, /// WhatsApp product/catalog message. Product, Event, @@ -95,6 +97,7 @@ impl MessageKind { Self::ViewOnce => "view_once", Self::Hosted => "hosted", Self::Bot => "bot", + Self::HistoryNotice => "history_notice", Self::Unknown => "unknown", Self::Other(label) => label, } @@ -140,6 +143,7 @@ impl MessageKind { "view_once" => Self::ViewOnce, "hosted" => Self::Hosted, "bot" => Self::Bot, + "history_notice" => Self::HistoryNotice, "unknown" => Self::Unknown, _ => Self::Other(label), } @@ -307,6 +311,8 @@ pub struct StoredMessage { pub revoked: bool, /// A locally created tombstone that never contributed to unread counts. pub local_revoke_placeholder: bool, + /// Shared history is read backfill and never contributes unread attention. + pub received_as_shared_history: bool, /// Arrival order within this store, ascending. Opaque: compare it, don't /// interpret it. It exists because the server's `t` is whole seconds, so /// two messages exchanged in the same second carry the same `timestamp` diff --git a/crates/chat-store/tests/regression_245.rs b/crates/chat-store/tests/regression_245.rs new file mode 100644 index 00000000..f9786656 --- /dev/null +++ b/crates/chat-store/tests/regression_245.rs @@ -0,0 +1,119 @@ +//! These regressions use only the pre-fix public API so they also run on base. +mod common; +use common::*; + +#[tokio::test] +async fn shared_history_carrier_is_not_a_timeline_message() { + let (db, store) = test_store().await; + let device = db.device_id(); + db.shared() + .run(move |conn| { + diesel::sql_query("UPDATE device SET pn = ? WHERE id = ?") + .bind::("559900000099@s.whatsapp.net") + .bind::(device) + .execute(conn) + .map_err(db_err)?; + Ok(()) + }) + .await + .unwrap(); + let message = { + let mut fixture: wa::Message = Default::default(); + let mut metadata: wa::message::MessageHistoryMetadata = Default::default(); + metadata.history_receivers = vec!["559900000099@s.whatsapp.net".into()]; + metadata.message_count = Some(1); + metadata.oldest_message_timestamp_in_window = Some(1_700_000_000); + let mut bundle: wa::message::MessageHistoryBundle = Default::default(); + bundle.mimetype = Some("application/protobuf".into()); + bundle.direct_path = Some("/synthetic-history".into()); + bundle.media_key = Some(vec![1; 32]); + bundle.file_sha256 = Some(vec![2; 32]); + bundle.file_enc_sha256 = Some(vec![3; 32]); + bundle.message_history_metadata = MessageField::some(metadata); + fixture.message_history_bundle = MessageField::some(bundle); + fixture + }; + feed( + &store, + [message_event( + message, + incoming_info(GROUP, PEER, "BUNDLE", 1_700_000_100), + )], + ) + .await; + assert!( + store + .messages(&jid(GROUP), None, 100) + .await + .unwrap() + .is_empty(), + "history carriers must never escape as unknown bubbles" + ); +} + +#[tokio::test] +async fn historical_copy_positions_a_revoke_that_arrived_first() { + let (_db, store) = test_store().await; + let revoke = { + let mut fixture: wa::Message = Default::default(); + fixture.protocol_message = MessageField::some({ + let mut fixture: wa::message::ProtocolMessage = Default::default(); + fixture.r#type = Some(wa::message::protocol_message::Type::REVOKE); + fixture.key = MessageField::some({ + let mut fixture: wa::MessageKey = Default::default(); + fixture.id = Some("TARGET".into()); + fixture.from_me = Some(true); + fixture + }); + fixture + }); + fixture + }; + feed( + &store, + [message_event( + revoke, + incoming_info(GROUP, PEER, "REVOKE", 1_700_000_100), + )], + ) + .await; + let wmi = { + let mut fixture: wa::WebMessageInfo = Default::default(); + fixture.key = MessageField::some({ + let mut fixture: wa::MessageKey = Default::default(); + fixture.id = Some("TARGET".into()); + fixture.from_me = Some(false); + fixture.participant = Some(PEER.into()); + fixture + }); + fixture.message_timestamp = Some(1_700_000_010); + fixture.message = MessageField::some(wa::Message::text("synthetic")); + fixture + }; + feed( + &store, + [history_sync_event({ + let mut fixture: wa::HistorySync = Default::default(); + fixture.sync_type = wa::history_sync::HistorySyncType::RECENT; + fixture.conversations = vec![{ + let mut fixture: wa::Conversation = Default::default(); + fixture.id = GROUP.into(); + fixture.messages = vec![{ + let mut fixture: wa::HistorySyncMsg = Default::default(); + fixture.message = MessageField::some(wmi); + fixture + }]; + fixture + }]; + fixture + })], + ) + .await; + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 1); + assert!(rows[0].revoked); + assert!(rows[0].message.is_none()); + assert!(rows[0].text.is_none()); + assert_eq!(rows[0].timestamp, ts(1_700_000_010)); + assert_eq!(rows[0].kind.as_str(), "text"); +} diff --git a/crates/chat-store/tests/shared_history.rs b/crates/chat-store/tests/shared_history.rs new file mode 100644 index 00000000..757c8525 --- /dev/null +++ b/crates/chat-store/tests/shared_history.rs @@ -0,0 +1,535 @@ +//! Synthetic pairwise bundle flows. No session, captures or personal messages. +#![allow(clippy::disallowed_methods)] +mod common; +use common::*; +use std::io::Write; +const OWN: &str = "559900000099@s.whatsapp.net"; +const OTHER: &str = "559900000002@s.whatsapp.net"; +const AT: i64 = 1_700_000_100; + +async fn own(store: &SqliteStore) { + let id = store.device_id(); + store + .shared() + .run(move |conn| { + diesel::sql_query("UPDATE device SET pn = ? WHERE id = ?") + .bind::(OWN) + .bind::(id) + .execute(conn) + .map_err(db_err)?; + Ok(()) + }) + .await + .unwrap(); +} +fn bundle(receivers: &[&str]) -> wa::Message { + { + let mut fixture: wa::Message = Default::default(); + fixture.message_history_bundle = MessageField::some({ + let mut fixture: wa::message::MessageHistoryBundle = Default::default(); + fixture.mimetype = Some("application/protobuf".into()); + fixture.direct_path = Some("/synthetic-history".into()); + fixture.media_key = Some(vec![1; 32]); + fixture.file_sha256 = Some(vec![2; 32]); + fixture.file_enc_sha256 = Some(vec![3; 32]); + fixture.message_history_metadata = MessageField::some({ + let mut fixture: wa::message::MessageHistoryMetadata = Default::default(); + fixture.history_receivers = receivers.iter().map(|s| s.to_string()).collect(); + fixture.message_count = Some(3); + fixture.oldest_message_timestamp_in_window = Some(AT - 90); + fixture.oldest_message_timestamp_in_bundle = Some(AT - 90); + fixture + }); + fixture + }); + fixture + } +} +fn historical(id: &str, author: Option<&str>, from_me: bool, time: i64) -> wa::WebMessageInfo { + { + let mut fixture: wa::WebMessageInfo = Default::default(); + fixture.key = MessageField::some({ + let mut fixture: wa::MessageKey = Default::default(); + fixture.remote_jid = Some(GROUP.into()); + fixture.id = Some(id.into()); + fixture.from_me = Some(from_me); + fixture.participant = author.map(str::to_owned); + fixture + }); + fixture.message_timestamp = Some(time as u64); + fixture.message = MessageField::some(wa::Message::text(id)); + fixture + } +} +fn compressed(messages: Vec) -> Vec { + use buffa::Message; + let history = { + let mut fixture: wa::GroupHistory = Default::default(); + fixture.messages = messages; + fixture + }; + let mut encoder = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(&history.encode_to_vec()).unwrap(); + encoder.finish().unwrap() +} +async fn receive(store: &ChatStore, message: wa::Message, id: &str) { + feed( + store, + [message_event(message, incoming_info(GROUP, PEER, id, AT))], + ) + .await; +} + +#[tokio::test] +async fn received_bundle_is_durable_work_without_an_unknown_bubble() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + assert!( + store + .messages(&jid(GROUP), None, 100) + .await + .unwrap() + .is_empty() + ); + let reopened = ChatStore::new(&db).await.unwrap(); + let pending = reopened.pending_history_bundles().await.unwrap(); + assert_eq!(pending.len(), 1); + let bytes = compressed(vec![ + historical("ADMIN", None, true, AT - 50), + historical("MINE", Some(OWN), false, AT - 60), + historical("THIRD", Some(OTHER), false, AT - 70), + ]); + reopened.apply_history_bundle(&pending[0], &bytes).unwrap(); + reopened.flush().await.unwrap(); + reopened.apply_history_bundle(&pending[0], &bytes).unwrap(); + reopened.flush().await.unwrap(); + receive(&reopened, bundle(&[OWN]), "BUNDLE").await; + assert!(reopened.pending_history_bundles().await.unwrap().is_empty()); + let rows = reopened.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!( + rows.iter().map(|r| r.id.as_str()).collect::>(), + ["ADMIN", "MINE", "THIRD"] + ); + assert_eq!(rows[0].sender_jid, jid(PEER)); + assert!(rows[1].from_me); + assert_eq!(rows[2].sender_jid, jid(OTHER)); + assert_eq!( + reopened + .chat(&jid(GROUP)) + .await + .unwrap() + .unwrap() + .unread_count, + 0 + ); +} + +#[tokio::test] +async fn unshared_own_and_sync_bundles_do_not_become_pending_work() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, bundle(&[OTHER]), "NOT-FOR-ME").await; + let mut info = incoming_info(GROUP, OWN, "OWN-BUNDLE", AT); + info.source.is_from_me = true; + feed(&store, [message_event(bundle(&[OWN]), info)]).await; + feed( + &store, + [history_sync_event({ + let mut fixture: wa::HistorySync = Default::default(); + fixture.sync_type = wa::history_sync::HistorySyncType::RECENT; + fixture.conversations = vec![{ + let mut fixture: wa::Conversation = Default::default(); + fixture.id = GROUP.into(); + fixture.messages = vec![{ + let mut fixture: wa::HistorySyncMsg = Default::default(); + fixture.message = MessageField::some({ + let mut fixture: wa::WebMessageInfo = Default::default(); + fixture.key = MessageField::some({ + let mut fixture: wa::MessageKey = Default::default(); + fixture.id = Some("SYNC".into()); + fixture + }); + fixture.message = MessageField::some(bundle(&[OWN])); + fixture + }); + fixture + }]; + fixture + }]; + fixture + })], + ) + .await; + assert!(store.pending_history_bundles().await.unwrap().is_empty()); + assert!( + store + .messages(&jid(GROUP), None, 100) + .await + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn malformed_partial_and_wrong_chat_bodies_cannot_leak_unknown_rows() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let pending = store.pending_history_bundles().await.unwrap().remove(0); + assert!(store.apply_history_bundle(&pending, b"not zlib").is_err()); + assert_eq!(store.pending_history_bundles().await.unwrap().len(), 1); + let valid = compressed(vec![historical("GOOD", Some(OTHER), false, AT - 10)]); + assert!( + store + .apply_history_bundle(&pending, &valid[..valid.len() - 3]) + .is_err() + ); + let mut trailing = valid.clone(); + trailing.extend_from_slice(b"trailing"); + assert!(store.apply_history_bundle(&pending, &trailing).is_err()); + + let mut wrong = historical("WRONG", Some(OTHER), false, AT - 10); + wrong.key.as_option_mut().unwrap().remote_jid = Some(PEER.into()); + let mut unknown = historical("UNKNOWN", Some(OTHER), false, AT - 10); + unknown.message = MessageField::some({ + let mut fixture: wa::Message = Default::default(); + fixture.message_context_info = MessageField::some(Default::default()); + fixture + }); + let bytes = compressed(vec![ + wrong, + unknown, + historical("INVALID-AUTHOR", Some("invalid"), false, AT - 10), + historical("OLD", Some(OTHER), false, AT - 100), + historical("FUTURE", Some(OTHER), false, AT + 1), + historical("GOOD", Some(OTHER), false, AT - 10), + ]); + store.apply_history_bundle(&pending, &bytes).unwrap(); + store.flush().await.unwrap(); + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].id, "GOOD"); + assert!( + store + .messages(&jid(PEER), None, 100) + .await + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn revoke_before_history_recovers_position_without_resurrecting_content() { + let (db, store) = test_store().await; + own(&db).await; + let revoke = { + let mut fixture: wa::Message = Default::default(); + fixture.protocol_message = MessageField::some({ + let mut fixture: wa::message::ProtocolMessage = Default::default(); + fixture.r#type = Some(wa::message::protocol_message::Type::REVOKE); + fixture.key = MessageField::some({ + let mut fixture: wa::MessageKey = Default::default(); + fixture.id = Some("DELETED".into()); + fixture.from_me = Some(true); + fixture.participant = Some(PEER.into()); + fixture + }); + fixture + }); + fixture + }; + receive(&store, revoke, "REVOKE").await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let pending = store.pending_history_bundles().await.unwrap().remove(0); + store + .apply_history_bundle( + &pending, + &compressed(vec![historical("DELETED", None, true, AT - 50)]), + ) + .unwrap(); + store.flush().await.unwrap(); + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 1); + assert!(rows[0].revoked); + assert!(rows[0].text.is_none()); + assert!(rows[0].message.is_none()); + assert_eq!(rows[0].kind, MessageKind::Text); + assert_eq!(rows[0].timestamp, ts(AT - 50)); +} + +#[tokio::test] +async fn inbound_durability_boundary_persists_work_before_returning() { + let (db, store) = test_store().await; + own(&db).await; + let inbound = InboundMessage::builder() + .message(Arc::new(bundle(&[OWN]))) + .info(Arc::new(incoming_info(GROUP, PEER, "HOOK", AT))) + .build(); + store.commit_inbound_batch(&[inbound]).await.unwrap(); + let reopened = ChatStore::new(&db).await.unwrap(); + assert_eq!(reopened.pending_history_bundles().await.unwrap().len(), 1); + assert!( + reopened + .messages(&jid(GROUP), None, 100) + .await + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn stale_history_preserves_live_content_and_same_id_other_authors() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, wa::Message::text("live copy"), "SAME").await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let pending = store.pending_history_bundles().await.unwrap().remove(0); + store + .apply_history_bundle( + &pending, + &compressed(vec![ + historical("SAME", None, true, AT - 50), + historical("SAME", Some(OTHER), false, AT - 60), + historical("SAME", Some(OTHER), false, AT - 60), + historical("IMPLICIT-OWN", None, false, AT - 65), + ]), + ) + .unwrap(); + store.flush().await.unwrap(); + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!(rows[0].text.as_deref(), Some("live copy")); + assert!( + rows.iter() + .find(|r| r.id == "IMPLICIT-OWN") + .unwrap() + .from_me + ); + store.reconcile_chat(&jid(GROUP)).unwrap(); + store.flush().await.unwrap(); + assert_eq!( + store.chat(&jid(GROUP)).await.unwrap().unwrap().unread_count, + 1 + ); +} + +#[tokio::test] +async fn failed_work_rotates_and_discard_survives_restart_and_redelivery() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, bundle(&[OWN]), "A").await; + receive(&store, bundle(&[OWN]), "B").await; + let pending = store.pending_history_bundles().await.unwrap(); + store.record_history_bundle_attempt(&pending[0]).unwrap(); + store.flush().await.unwrap(); + assert_eq!( + store.pending_history_bundles().await.unwrap()[0].msg_id, + "B" + ); + store.discard_history_bundle(&pending[0]).unwrap(); + store.flush().await.unwrap(); + let reopened = ChatStore::new(&db).await.unwrap(); + receive(&reopened, bundle(&[OWN]), "A").await; + assert_eq!(reopened.pending_history_bundles().await.unwrap().len(), 1); + assert_eq!( + reopened.pending_history_bundles().await.unwrap()[0].msg_id, + "B" + ); +} + +#[tokio::test] +async fn classifier_repair_moves_legacy_carriers_and_corrects_badges() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, wa::Message::text("live"), "LIVE").await; + let device = db.device_id(); + let proto = waproto::codec::message_to_vec(&bundle(&[OWN])); + db.shared().run(move |conn| { + diesel::sql_query("INSERT INTO messages (device_id, chat_jid, msg_id, sender_jid, from_me, timestamp_ms, kind, proto) VALUES (?, ?, 'LEGACY', ?, FALSE, ?, 'unknown', ?)") + .bind::(device).bind::(GROUP) + .bind::(PEER).bind::((AT+1)*1000) + .bind::(proto).execute(conn).map_err(db_err)?; + diesel::sql_query("UPDATE chats SET unread_count = 2, last_message_ts = ? WHERE device_id = ? AND jid = ?") + .bind::((AT+1)*1000).bind::(device) + .bind::(GROUP).execute(conn).map_err(db_err)?; + diesel::sql_query("UPDATE chat_store_meta SET value = '2026-09-24-v1' WHERE key = 'message_kind_classifier'").execute(conn).map_err(db_err)?; + Ok(()) + }).await.unwrap(); + ChatStore::prepare(&db).await.unwrap(); + assert_eq!(store.pending_history_bundles().await.unwrap().len(), 1); + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 1); + let chat = store.chat(&jid(GROUP)).await.unwrap().unwrap(); + assert_eq!(chat.unread_count, 1); + ChatStore::prepare(&db).await.unwrap(); + assert_eq!(store.pending_history_bundles().await.unwrap().len(), 1); +} + +#[tokio::test] +async fn notices_are_system_rows_only_for_nonreceivers_and_never_unread() { + let (db, store) = test_store().await; + own(&db).await; + let notice = |receivers: &[&str]| { + let mut metadata: wa::message::MessageHistoryMetadata = Default::default(); + metadata.history_receivers = receivers.iter().map(|s| s.to_string()).collect(); + metadata.message_count = Some(3); + metadata.oldest_message_timestamp_in_window = Some(AT - 90); + let mut notice: wa::message::MessageHistoryNotice = Default::default(); + notice.message_history_metadata = MessageField::some(metadata); + let mut message: wa::Message = Default::default(); + message.message_history_notice = MessageField::some(notice); + message + }; + receive(&store, notice(&[OWN]), "RECIPIENT-NOTICE").await; + receive(&store, notice(&[OTHER]), "VISIBLE-NOTICE").await; + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].kind, MessageKind::HistoryNotice); + assert!( + rows[0] + .text + .as_deref() + .unwrap() + .contains("3 earlier messages") + ); + assert_eq!( + store.chat(&jid(GROUP)).await.unwrap().unwrap().unread_count, + 0 + ); + store.reconcile_chat(&jid(GROUP)).unwrap(); + store.flush().await.unwrap(); + assert_eq!( + store.chat(&jid(GROUP)).await.unwrap().unwrap().unread_count, + 0 + ); + feed( + &store, + [message_event( + wa::Message::text("older unread"), + incoming_info(GROUP, PEER, "UNREAD", AT - 10), + )], + ) + .await; + let (head, _) = store.page_with_unread(&jid(GROUP), None, 1).await.unwrap(); + let (older, unread) = store + .page_with_unread(&jid(GROUP), Some((&head[0]).into()), 1) + .await + .unwrap(); + assert_eq!(older[0].id, "UNREAD"); + assert_eq!( + unread, 1, + "the newer system notice cannot spend an unread slot" + ); +} + +#[tokio::test] +async fn invalid_envelopes_and_restricted_or_expired_content_are_not_injected() { + let (db, store) = test_store().await; + own(&db).await; + let mut invalid = bundle(&[OWN]); + invalid + .message_history_bundle + .as_option_mut() + .unwrap() + .media_key = Some(vec![0; 1]); + receive(&store, invalid, "INVALID").await; + assert!(store.pending_history_bundles().await.unwrap().is_empty()); + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let pending = store.pending_history_bundles().await.unwrap().remove(0); + let mut view_once = historical("VIEW-ONCE", Some(OTHER), false, AT - 10); + let mut image: wa::message::ImageMessage = Default::default(); + image.view_once = Some(true); + let mut content: wa::Message = Default::default(); + content.image_message = MessageField::some(image); + view_once.message = MessageField::some(content); + let mut expired = historical("EXPIRED", Some(OTHER), false, AT - 10); + let mut text: wa::message::ExtendedTextMessage = Default::default(); + let mut context: wa::ContextInfo = Default::default(); + context.expiration = Some(1); + text.context_info = MessageField::some(context); + text.text = Some("synthetic expired".into()); + let mut content: wa::Message = Default::default(); + content.extended_text_message = MessageField::some(text); + expired.message = MessageField::some(content); + store + .apply_history_bundle(&pending, &compressed(vec![view_once, expired])) + .unwrap(); + store.flush().await.unwrap(); + assert!( + store + .messages(&jid(GROUP), None, 100) + .await + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn pending_and_consumed_work_survive_closing_every_database_handle() { + let path = std::env::temp_dir().join(format!( + "oxidezap-shared-history-restart-{}.db", + std::process::id() + )); + let path_str = path.to_str().unwrap(); + let db = SqliteStore::open(path_str).await.unwrap(); + db.database() + .provision_device(db.device_id()) + .await + .unwrap(); + own(&db).await; + let store = ChatStore::new(&db).await.unwrap(); + receive(&store, bundle(&[OWN]), "RESTART").await; + store.close().await.unwrap(); + drop(store); + drop(db); + let db = SqliteStore::open(path_str).await.unwrap(); + let store = ChatStore::new(&db).await.unwrap(); + let pending = store.pending_history_bundles().await.unwrap().remove(0); + store + .apply_history_bundle( + &pending, + &compressed(vec![historical("PERSISTED", None, true, AT - 10)]), + ) + .unwrap(); + store.close().await.unwrap(); + drop(store); + drop(db); + let db = SqliteStore::open(path_str).await.unwrap(); + let store = ChatStore::new(&db).await.unwrap(); + assert!(store.pending_history_bundles().await.unwrap().is_empty()); + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].id, "PERSISTED"); + store.close().await.unwrap(); + drop(store); + drop(db); + for suffix in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{path_str}{suffix}")); + } +} + +#[tokio::test] +async fn multi_chunk_inflation_preserves_content_and_rejects_invalid_protobuf() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let pending = store.pending_history_bundles().await.unwrap().remove(0); + let mut encoder = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(&[0xff]).unwrap(); + assert!( + store + .apply_history_bundle(&pending, &encoder.finish().unwrap()) + .is_err() + ); + let mut message = historical("LARGE", Some(OTHER), false, AT - 10); + message.message = MessageField::some(wa::Message::text("s".repeat(30_000))); + store + .apply_history_bundle(&pending, &compressed(vec![message])) + .unwrap(); + store.flush().await.unwrap(); + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows[0].text.as_deref().unwrap().len(), 30_000); +} diff --git a/crates/session/src/whatsapp/convert.rs b/crates/session/src/whatsapp/convert.rs index b63fcac8..4731b898 100644 --- a/crates/session/src/whatsapp/convert.rs +++ b/crates/session/src/whatsapp/convert.rs @@ -28,7 +28,10 @@ pub(super) fn stored_with_unread_tail( .into_iter() .rev() .map(|row| { - let eligible = !row.from_me && !row.local_revoke_placeholder; + let eligible = !row.from_me + && !row.local_revoke_placeholder + && !row.received_as_shared_history + && row.kind != oxidezap_chat_store::MessageKind::HistoryNotice; let mut message = stored_to_chat_message(row); if eligible && remaining > 0 { message.is_read = false; @@ -77,6 +80,8 @@ pub(super) fn stored_to_chat_message(stored: oxidezap_chat_store::StoredMessage) true }; let quoted = base_message.and_then(quoted_from); + let system = (stored.kind == oxidezap_chat_store::MessageKind::HistoryNotice) + .then(|| oxidezap_core::SystemNotice::GroupChanged(content.clone())); ChatMessage { id: stored.id, sender: stored.sender_jid.to_string(), @@ -99,7 +104,7 @@ pub(super) fn stored_to_chat_message(stored: oxidezap_chat_store::StoredMessage) quoted, revoked: stored.revoked, edited: stored.edited_at.is_some(), - system: None, + system, poll, } } @@ -279,10 +284,32 @@ mod tests { edited_at: None, revoked: false, local_revoke_placeholder: false, + received_as_shared_history: false, seq: 1, } } + #[test] + fn shared_history_is_read_and_not_an_unknown_system_bubble() { + let mut historical = stored_poll_creation(); + historical.received_as_shared_history = true; + let mut live = stored_poll_creation(); + live.id = "LIVE".into(); + let messages = stored_with_unread_tail(vec![live, historical], 1); + assert!(!messages[0].is_read); + assert!(messages[1].is_read); + let mut notice = stored_poll_creation(); + notice.kind = oxidezap_chat_store::MessageKind::HistoryNotice; + notice.text = Some("Shared earlier messages with group members".into()); + notice.message = None; + let rendered = stored_to_chat_message(notice); + assert!(matches!( + rendered.system, + Some(oxidezap_core::SystemNotice::GroupChanged(_)) + )); + assert!(!rendered.content.contains("unknown")); + } + #[test] fn local_placeholder_does_not_consume_the_hydrated_unread_tail() { let mut incoming = stored_poll_creation(); diff --git a/crates/session/src/whatsapp/mod.rs b/crates/session/src/whatsapp/mod.rs index 065788c8..0b0b9aa4 100644 --- a/crates/session/src/whatsapp/mod.rs +++ b/crates/session/src/whatsapp/mod.rs @@ -31,6 +31,7 @@ mod sends; mod history; /// Which events wait for which, and which never had to. mod lanes; +mod shared_history; /// What a message's media is, for both the roads it arrives on. mod media; @@ -1461,6 +1462,8 @@ impl WhatsAppClient { }); } + shared_history::spawn(bot.client(), chat_store.clone(), stopping.clone()); + // Re-hydrate the UI off the store's invalidation stream instead of raw // client events: changes are emitted only after commit, so a reload can // never observe pre-commit state (no flush barrier, no dispatch-order @@ -2226,7 +2229,7 @@ impl WhatsAppClient { // is a `[Media]` bubble with an unread badge, and the store writes no // row for it, so it vanishes at the next hydration having already // raised a count nothing sent a receipt for. - if oxidezap_chat_store::is_control_only(msg) { + if oxidezap_chat_store::is_control_only(msg) || base_msg.message_history_notice.is_set() { return; } diff --git a/crates/session/src/whatsapp/shared_history.rs b/crates/session/src/whatsapp/shared_history.rs new file mode 100644 index 00000000..695fdeb0 --- /dev/null +++ b/crates/session/src/whatsapp/shared_history.rs @@ -0,0 +1,127 @@ +//! Download persisted attachments serially. Failures stay pending across restarts. +use oxidezap_chat_store::{ChatStore, ChatStoreError}; +use std::{sync::Arc, time::Duration}; +use whatsapp_rust::{ + client::Client, + download::{ClientDownloadError, DownloadLimits, DownloadPolicyError}, + error::ErrorChainExt, +}; + +pub(super) fn spawn( + client: Arc, + store: Arc, + mut stopping: tokio::sync::watch::Receiver<()>, +) { + crate::exec::spawn_owned(async move { + loop { + // One bounded page and one network request at a time. Cancellation + // drops only in-flight work; the durable attachment remains pending. + tokio::select! { + _ = stopping.changed() => break, + _ = pass(&client, &store) => {} + } + tokio::select! { + _ = stopping.changed() => break, + _ = crate::exec::sleep(Duration::from_secs(30)) => {} + } + } + }); +} + +async fn pass(client: &Arc, store: &ChatStore) { + let bundles = match store.pending_history_bundles().await { + Ok(bundles) => bundles, + Err(error) => { + log::warn!("shared history queue read failed: {error}"); + return; + } + }; + for pending in bundles { + if let Err(error) = store.record_history_bundle_attempt(&pending) { + log::warn!("shared history attempt admission failed: {error}"); + break; + } + let Ok(message) = pending.message() else { + if let Err(error) = store.discard_history_bundle(&pending) { + log::warn!("shared history discard failed: {error}"); + } + continue; + }; + let Some(bundle) = oxidezap_chat_store::normalized_message(&message) + .message_history_bundle + .as_option() + else { + if let Err(error) = store.discard_history_bundle(&pending) { + log::warn!("shared history discard failed: {error}"); + } + continue; + }; + let result = tokio::select! { + result = client.download(bundle, DownloadLimits::new(32 * 1024 * 1024, 32 * 1024 * 1024)) => result, + _ = crate::exec::sleep(Duration::from_secs(60)) => { + log::warn!("shared history download timed out; attachment remains pending"); + continue; + } + }; + match result { + Ok(bytes) => { + if let Err(error) = store.apply_history_bundle(&pending, &bytes) { + log::warn!("shared history application failed: {error}"); + if terminal_body_error(&error) + && let Err(error) = store.discard_history_bundle(&pending) + { + log::warn!("shared history discard failed: {error}"); + } + } + } + Err(error) if terminal_download_error(&error) => { + if let Err(error) = store.discard_history_bundle(&pending) { + log::warn!("shared history discard failed: {error}"); + } + } + Err(_) => { + // No message bodies, media paths or cryptographic keys in logs. + log::warn!("shared history download failed; attachment remains pending"); + } + } + } + if let Err(error) = store.flush().await { + log::warn!("shared history commit failed: {error}"); + } +} + +fn terminal_download_error(error: &ClientDownloadError) -> bool { + matches!( + error, + ClientDownloadError::ResourcePolicy(DownloadPolicyError::LimitExceeded { .. }) + ) || error + .http_status() + .is_some_and(|status| matches!(status, 404 | 410)) +} + +fn terminal_body_error(error: &ChatStoreError) -> bool { + matches!(error, ChatStoreError::InvalidGroupHistory(_)) +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn writer_pressure_keeps_downloaded_history_retryable() { + assert!(!terminal_download_error( + &ClientDownloadError::ResourcePolicy(DownloadPolicyError::StreamingRequired) + )); + assert!(terminal_download_error( + &ClientDownloadError::ResourcePolicy(DownloadPolicyError::LimitExceeded { + domain: whatsapp_rust::download::MediaByteDomain::Wire, + limit: 32, + }) + )); + assert!(!terminal_body_error(&ChatStoreError::WriteBatchFailed( + "queue full".into() + ))); + assert!(terminal_body_error(&ChatStoreError::InvalidGroupHistory( + "invalid zlib".into() + ))); + } +} diff --git a/crates/session/src/whatsapp/tests.rs b/crates/session/src/whatsapp/tests.rs index 9b1337fd..b029dcbc 100644 --- a/crates/session/src/whatsapp/tests.rs +++ b/crates/session/src/whatsapp/tests.rs @@ -2471,6 +2471,7 @@ fn a_message_cursor_survives_the_round_trip() { status: oxidezap_chat_store::MessageStatus::Delivered, revoked: false, local_revoke_placeholder: false, + received_as_shared_history: false, edited_at: None, starred: false, seq: 4242, @@ -2517,6 +2518,7 @@ fn stored_edit_state_reaches_own_and_peer_bubbles_in_direct_and_group_chats() { edited_at: Some(timestamp), revoked: false, local_revoke_placeholder: false, + received_as_shared_history: false, seq: 1, }; let mut plain = stored.clone(); diff --git a/docs/shared-group-history.md b/docs/shared-group-history.md new file mode 100644 index 00000000..6ccb7052 --- /dev/null +++ b/docs/shared-group-history.md @@ -0,0 +1,146 @@ +# Receiving deliberately shared group history + +Investigation for [client #245](https://github.com/oxidezap/client/issues/245), +2026-10-22. Static protocol analysis and synthetic SQLite tests only. No account +was connected, and the issue author's authenticated observation was not repeated. + +## Evidence snapshot + +The remote whatspec main was re-fetched and resolved to +`46f630441389e703e04ff3fc012192cf80fdd4ba`. Manifest and bundle lock were read +from that commit, rather than the older local checkout. Schema is `4.4.0`, +WhatsApp Web snapshot is `2.3000.1047483476`, with 579 bundles and set hash +`05609307e68b0f6ccfd2a121a573049999e38b752cfe7155ebbe5c661805751b`. +This identifies preserved bytes, not today's deployed WhatsApp service. + +The [exact release asset](https://github.com/oxidezap/whatspec/releases/download/bundle-store/bundles-2.3000.1047483476-05609307e68b0f6ccfd2a121a573049999e38b752cfe7155ebbe5c661805751b.tar.xz) +was already cached. Its SHA-256 is +`22243b9c2ff18a66cea8fd66551821fe0f9f09786d107e2be8284637db86c349`, +matching the release asset digest. All 579 individual sizes and SHA-256 hashes +were checked against the selected lock. Archive hash and set hash are distinct. + +Acorn parsed the original JavaScript without executing it. Complete `__d` +definitions were extracted with byte offsets. The first history-focused extraction found 112 definitions with 82 unique +names. All 30 duplicate names have identical factory hashes. Additional parser +registry and validation modules were then followed through their imports. +Relevant modules and half-open byte ranges are below. + +| Module | Bundle SHA-256 | Bytes | +| --- | --- | --- | +| `WAWebDecompressAndDecodeBundle` | `012a3b924a76e250362922ed59882da7bfe95b399d33d3da1426e90323d10e5a` | 936551..936924 | +| `WAWebMessageHistoryValidation` | `09d91fdc4089c50ab1d35dd6c796ee82ef4c3296710316d20383c85738b871b8` | 4442707..4446708 | +| `WAWebGroupHistoryBundleProcessor` | `012a3b924a76e250362922ed59882da7bfe95b399d33d3da1426e90323d10e5a` | 950427..967054 | +| `WAWebParseMessageHistoryBundleProto` | `09d91fdc4089c50ab1d35dd6c796ee82ef4c3296710316d20383c85738b871b8` | 4488598..4492768 | +| `WAWebParseMessageHistoryNoticeProto` | `09d91fdc4089c50ab1d35dd6c796ee82ef4c3296710316d20383c85738b871b8` | 4492770..4494956 | +| `WAWebGroupHistoryNoticeHandler` | `09d91fdc4089c50ab1d35dd6c796ee82ef4c3296710316d20383c85738b871b8` | 607474..610912 | +| `WAWebE2EProtoUtils` | `430cb8979511aada95178605faa35a8483446f677ce6d1fe0fd66333c7879eee` | 442807..459797 | +| `WAWebGroupHistoryUtils` | `f588ee2f033b87e018a3be34a37ec1347d8b6b8700de539e6be31481b29c4690` | 4362646..4364221 | +| `WAWebGenerateMessageHistoryBundleProto` | `f588ee2f033b87e018a3be34a37ec1347d8b6b8700de539e6be31481b29c4690` | 1299976..1300973 | +| `WAWebGenerateMessageHistoryNoticeProto` | `f588ee2f033b87e018a3be34a37ec1347d8b6b8700de539e6be31481b29c4690` | 1300975..1301605 | + +The [manifest](https://github.com/oxidezap/whatspec/blob/46f630441389e703e04ff3fc012192cf80fdd4ba/generated/manifest.json) +and [lock](https://github.com/oxidezap/whatspec/blob/46f630441389e703e04ff3fc012192cf80fdd4ba/generated/bundles.lock.json) +are the reproducible index. Diagnostics include incoming drops for two nonliteral +attributes and seven shadowed callback reads. Notification diagnostics report five degraded shapes and one drop each for +unresolved action tables, nonliteral attributes, byte content and shadowed +callback reads. The incoming/notification schemas were inspected too. Absence from those catalogs alone is not evidence +that a handler does not exist. + +## Observed behavior and implementation boundary + +The message parser plugin registry registers the two history parsers as +separate message plugins. `WAWebMessageHistoryValidation` requires nonempty +receivers, an oldest-window timestamp, a positive count, protobuf MIME type +and 32-byte media key and hashes. `WAWebDecompressAndDecodeBundle` calls +`WAGzip.inflate` before the group-history protobuf decoder. + +The Web parses a media `messageHistoryBundle` separately from the group +`messageHistoryNotice`. `WAWebGroupHistoryUtils.isValidHistoryBundleToProcess` +requires a received group bundle addressed to the local account in +`historyReceivers`, as well as process-state and time-window guards. The notice +parser hides the notice from the sharing account and history recipients and +shows it to other group members. The notice handler updates per-participant +sharing state; it does not contain the shared message bodies. + +The bundle processor inflates and decodes `GroupHistory`, translates message +keys from the sharing account's perspective, validates chat ownership, skips +expired/unsupported messages and existing rows, and injects the accepted rows. +`translateRegularMessageKeyToLocalReference` calls `W` in `WAWebE2EProtoUtils`. +For an incoming share, a true `fromMe` refers to the sharing account. A false +bit with no participant refers to the recipient's own account. This differs +from copying `WebMessageInfo` directly into ordinary local history. + +The processor also handles reporting tokens, out-of-window pins, add-ons and +receiver feature flags. This correction does not claim complete parity with +those Web features or infer a server contract from module names. In particular, +only the primary `messages` collection is materialized here. No request for +unshared history, Restricted-chat access, view-once recovery or account-policy +change is added. + +The current remote engine was `ac9d0df3d6269c6b5b60881a7154414d7c348991`; +the bridge was `dc96100911d99a6076c521965e47763f1c6123a2`. Their current source +contains send/retry support and codecs, but no receiver that downloads a +`messageHistoryBundle` and emits its messages as history. The client still pins +all engine crates together at `633f01480fe3a95510172c6312d01484563dcf3c`. +That version already implements `Downloadable` with the `Group History` HKDF +context and `/mms/group-history`, and `group_history_decode`. No dependency +update or generated protocol edits are needed. + +The group's notification metadata (`group_history_sent_state` on membership +updates) is independent of the pairwise message attachment. It already reaches +`group_notice.rs`; treating it as a download source would invent a contract. + +## Client behavior + +Inbound durability commits persist a pending attachment instead of an unknown +message. Own, unshared, nongroup and history-sync carrier copies create no work. +Completed/discarded keys remain as deduplication records; media envelopes +are cleared on consumption. A serial session +worker downloads bounded pages, keeps transient failures for retry, rotates +attempted work behind untouched work, and retires permanently missing media or +invalid bodies. Teardown cancels its in-flight request without losing work. + +A bounded, checksum-valid zlib stream and protobuf decode must complete before +any message insertion. Partial collections skip invalid individual entries. +Consumption and insertion share the writer transaction; history fills gaps, +preserves live edits/revokes and keeps sender identity. Shared rows retain a +persistent read-backfill marker so later unread recounts do not badge them. +Accepted rows keep their original timestamps and media protobufs. Store +invalidations hydrate the public UI after commit. Visible notices have their own +kind and render as group system lines, without unread attention. + +A versioned startup repair moves old unknown carriers to the pending table, +reclassifies visible notices, removes hidden notices and corrects carrier badge +contributions. When a sender-matched revoke placeholder precedes its content, +the later copy supplies only its original timestamp and kind. Deleted content +stays absent. + +Local ceilings of 32 MiB compressed/inflated and 10,000 top-level records bound +work; these are application policy, not asserted server limits. Effective Web +feature flags, reporting-token behavior, authenticated downloads and cross-device +server timing remain outside the synthetic tests. + +## Validation + +Local final validation used the normal Cargo cache with two build jobs: + +- `cargo fmt --all -- --check` passed. +- `cargo clippy -p oxidezap-chat-store -p oxidezap-session --all-targets --all-features -j 2 -- -D warnings` passed. +- `cargo nextest run -p oxidezap-chat-store -p oxidezap-session --all-features --build-jobs 2 --test-threads 2` passed all 672 tests, with two intentional skips. +- Both crates' doctests completed without failures, with three ignored examples. + +The exact same `tests/regression_245.rs`, SHA-256 +`dccce7af6dcf23e194bcb6ebd29e4395194ea601045fd1566d3ca8b95f36cc5c`, +ran on unchanged client base `0ba4118bcdb02c613fda94c5a0208bff44316e82`. +Both tests failed there: a valid received carrier became an unknown row, and a +revoke placeholder retained the revoke's timestamp. Both pass with the correction. +The other fixtures exercise durability-hook completion, complete database closure +and reopening, repeated and reordered copies, sender collisions, local identities, +malformed/truncated bodies, partial acceptance, expired and view-once content, +notices, unread pagination, queue pressure and legacy repair. + +An earlier combined `cargo test` invocation passed 281 session tests but exhausted +`a_session_is_never_observed_half_open`'s existing 500,000-poll opening budget. +Its partial-state assertion did not fail. That test's opening path precedes the +new downloader task, and it passed in both complete Nextest runs, including the +final run after the pagination correction. No change to that test was made.