diff --git a/crates/chat-store/src/store/chat_rows.rs b/crates/chat-store/src/store/chat_rows.rs index fa0a53be..9f7fb9a7 100644 --- a/crates/chat-store/src/store/chat_rows.rs +++ b/crates/chat-store/src/store/chat_rows.rs @@ -408,6 +408,7 @@ pub(super) fn delete_chat_rows( bound: Option<&RangeBound>, ) -> QueryResult<()> { use schema::messages::dsl as m; + super::shared_history::retire_for_deletion(conn, device_id, chat, bound)?; // Every branch below deletes from the same set — this chat's rows, minus // the starred ones when they are spared — and differs only in the time // window it adds. A boxed statement is consumed by `execute`, and the diff --git a/crates/chat-store/src/store/edit.rs b/crates/chat-store/src/store/edit.rs index 5050bf66..a0422622 100644 --- a/crates/chat-store/src/store/edit.rs +++ b/crates/chat-store/src/store/edit.rs @@ -16,6 +16,8 @@ use crate::store::writer::ChangeSet; /// edited content up front — `insert_message` skips edited rows, so the /// original's later arrival can't show pre-edit text. Returns whether the /// chat-list preview changed. +/// `shared` marks only a newly inserted placeholder as read backfill. An +/// existing live target keeps its original unread eligibility. #[allow(clippy::too_many_arguments)] pub(super) fn apply_edit( conn: &mut SqliteConnection, @@ -28,6 +30,7 @@ pub(super) fn apply_edit( new_kind: &str, new_proto: &[u8], ts_ms: i64, + shared: bool, changes: &mut ChangeSet, ) -> QueryResult { use schema::messages::dsl; @@ -84,6 +87,7 @@ pub(super) fn apply_edit( wa::web_message_info::Status::DELIVERY_ACK as i32 }), dsl::edited_at_ms.eq(Some(ts_ms)), + dsl::received_as_shared_history.eq(shared), )) .on_conflict_do_nothing() .execute(conn)? @@ -102,7 +106,7 @@ pub(super) fn apply_edit( ts_ms, preview: new_text, kind: Some(new_kind), - unread_delta: i32::from(!from_me), + unread_delta: i32::from(!from_me && !shared), }, )?; Ok(true) diff --git a/crates/chat-store/src/store/history_sync.rs b/crates/chat-store/src/store/history_sync.rs index 9971c43b..ae220bd0 100644 --- a/crates/chat-store/src/store/history_sync.rs +++ b/crates/chat-store/src/store/history_sync.rs @@ -437,6 +437,7 @@ pub(super) fn apply_history_message( new_kind, &new_proto, sender_timestamp_ms.unwrap_or(ts_ms), + shared, cs, )? { cs.chats = true; @@ -456,6 +457,7 @@ pub(super) fn apply_history_message( target_from_me, ts_ms, false, + shared, cs, )? { cs.chats = true; diff --git a/crates/chat-store/src/store/inbound.rs b/crates/chat-store/src/store/inbound.rs index c142ea14..cc5e6de9 100644 --- a/crates/chat-store/src/store/inbound.rs +++ b/crates/chat-store/src/store/inbound.rs @@ -138,6 +138,7 @@ pub(super) fn apply_inbound( new_kind, &new_proto, sender_timestamp_ms.unwrap_or(ts_ms), + false, cs, )? { cs.chats = true; @@ -180,6 +181,7 @@ pub(super) fn apply_inbound( target_from_me, ts_ms, false, + false, cs, )? { cs.chats = true; diff --git a/crates/chat-store/src/store/revoke.rs b/crates/chat-store/src/store/revoke.rs index 3f5fe9c4..b6639a74 100644 --- a/crates/chat-store/src/store/revoke.rs +++ b/crates/chat-store/src/store/revoke.rs @@ -19,6 +19,8 @@ use crate::store::writer::ChangeSet; /// placeholder's `status` to the column default where the edit picks one by /// authorship, and previews as nothing rather than as new text. What is left /// to share is the `bump_chat` call, which is already a function. +/// `shared` marks only a newly inserted placeholder as read backfill. An +/// existing live target keeps its original unread eligibility. #[allow(clippy::too_many_arguments)] pub(super) fn apply_revoke( conn: &mut SqliteConnection, @@ -29,6 +31,7 @@ pub(super) fn apply_revoke( target_from_me: bool, ts_ms: i64, locally_sent: bool, + shared: bool, changes: &mut ChangeSet, ) -> QueryResult { use schema::messages::dsl; @@ -77,6 +80,7 @@ pub(super) fn apply_revoke( // arrived, retain that origin separately from the target author // so future badge recounts cannot turn it into unread attention. dsl::local_revoke_placeholder.eq(locally_sent), + dsl::received_as_shared_history.eq(shared), )) .on_conflict_do_nothing() .execute(conn)? @@ -107,7 +111,7 @@ pub(super) fn apply_revoke( ts_ms, preview: None, kind: None, - unread_delta: i32::from(!target_from_me && !locally_sent), + unread_delta: i32::from(!target_from_me && !locally_sent && !shared), }, row_id, )?; diff --git a/crates/chat-store/src/store/shared_history.rs b/crates/chat-store/src/store/shared_history.rs index 91fac4dc..eb100c65 100644 --- a/crates/chat-store/src/store/shared_history.rs +++ b/crates/chat-store/src/store/shared_history.rs @@ -154,6 +154,17 @@ pub(super) fn persist( Ok(()) } +// Compare the complete envelope: a stale completion cannot consume a different +// attachment when a sender reuses its stanza id. +fn complete( + conn: &mut SqliteConnection, + device: i32, + bundle: &PendingHistoryBundle, +) -> QueryResult { + 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) +} + pub(super) fn apply( conn: &mut SqliteConnection, device: i32, @@ -161,11 +172,7 @@ pub(super) fn apply( 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 { + if complete(conn, device, bundle)? == 0 { return Ok(()); } let Some(history) = history else { @@ -271,6 +278,29 @@ pub(super) fn apply( if crate::message_kind(content) == "unknown" && content.protocol_message.is_unset() { continue; } + // Aggregated reaction keys use the same sharing-account perspective + // as the enclosing history message. Missing/invalid keys have no author. + message.reactions.retain_mut(|reaction| { + let Some(key) = reaction.key.as_option_mut() else { + return false; + }; + let author = if key.from_me == Some(true) { + Some(bundle.sender_jid.as_str()) + } else { + 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 { + return false; + }; + let author = stored_sender(author, false); + key.from_me = Some(own.contains(&author)); + key.participant = Some(author); + true + }); super::history_sync::apply_history_message( conn, device, @@ -437,3 +467,29 @@ pub(super) fn repair_legacy( .bind::(chat).bind::(ts).execute(conn)?; Ok(true) } + +/// Retire covered attachment facts in the same transaction as chat deletion. +/// Keep their keys so a redelivery or an in-flight completion cannot revive them. +pub(super) fn retire_for_deletion( + conn: &mut SqliteConnection, + device: i32, + chat: &str, + bound: Option<&super::read_state::RangeBound>, +) -> QueryResult<()> { + let pending = diesel::sql_query("SELECT chat_jid, msg_id, sender_jid, timestamp_ms, proto FROM shared_group_history WHERE device_id = ? AND chat_jid = ? AND completed = FALSE") + .bind::(device).bind::(chat).load::(conn)?; + for bundle in pending { + let covered = bound.is_none_or(|bound| { + bundle.timestamp_ms < bound.second_start_ms + || (bundle.timestamp_ms <= bound.second_end_ms + && bound + .keys + .as_ref() + .is_none_or(|keys| keys.contains(&bundle.msg_id))) + }); + if covered { + complete(conn, device, &bundle)?; + } + } + Ok(()) +} diff --git a/crates/chat-store/src/store/writer.rs b/crates/chat-store/src/store/writer.rs index d1d409b9..41400293 100644 --- a/crates/chat-store/src/store/writer.rs +++ b/crates/chat-store/src/store/writer.rs @@ -496,6 +496,7 @@ fn apply_writer_msg( kind, proto, *timestamp_ms, + false, cs, )? { cs.chats = true; @@ -520,6 +521,7 @@ fn apply_writer_msg( *target_from_me, *timestamp_ms, true, + false, cs, )? { cs.chats = true; diff --git a/crates/chat-store/tests/shared_history.rs b/crates/chat-store/tests/shared_history.rs index 757c8525..8d0dc5a8 100644 --- a/crates/chat-store/tests/shared_history.rs +++ b/crates/chat-store/tests/shared_history.rs @@ -533,3 +533,288 @@ async fn multi_chunk_inflation_preserves_content_and_rejects_invalid_protobuf() let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); assert_eq!(rows[0].text.as_deref().unwrap().len(), 30_000); } + +async fn review_245_deleted_pending_import(delete: bool) { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, wa::Message::text("existing"), "LIVE").await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let pending = store.pending_history_bundles().await.unwrap().remove(0); + let other_chat = "120363000000000002@g.us"; + feed( + &store, + [message_event( + bundle(&[OWN]), + incoming_info(other_chat, PEER, "BUNDLE", AT), + )], + ) + .await; + let event = if delete { + Event::DeleteChatUpdate( + wacore::types::events::DeleteChatUpdate::builder() + .jid(jid(GROUP)) + .delete_media(false) + .timestamp(ts(AT + 1)) + .action(Box::new(wa::sync_action_value::DeleteChatAction::default())) + .from_full_sync(false) + .build(), + ) + } else { + Event::ClearChatUpdate( + wacore::types::events::ClearChatUpdate::builder() + .jid(jid(GROUP)) + .delete_starred(false) + .delete_media(false) + .timestamp(ts(AT + 1)) + .action(Box::new(wa::sync_action_value::ClearChatAction::default())) + .from_full_sync(false) + .build(), + ) + }; + feed(&store, [event]).await; + store.close().await.unwrap(); + let store = ChatStore::new(&db).await.unwrap(); + let still_pending = store + .pending_history_bundles() + .await + .unwrap() + .into_iter() + .filter(|b| b.chat_jid == GROUP) + .count(); + store + .apply_history_bundle( + &pending, + &compressed(vec![historical("OLD", Some(OTHER), false, AT - 10)]), + ) + .unwrap(); + store.flush().await.unwrap(); + assert!( + store + .messages(&jid(GROUP), None, 100) + .await + .unwrap() + .is_empty(), + "late completion restored cleared history" + ); + if delete { + assert!( + store.chat(&jid(GROUP)).await.unwrap().is_none(), + "late completion recreated the deleted chat" + ); + } + assert_eq!( + still_pending, 0, + "deletion must retire durable pending work" + ); + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let remaining = store.pending_history_bundles().await.unwrap(); + assert_eq!( + remaining.len(), + 1, + "redelivery revived retired work or another chat was retired" + ); + assert_eq!(remaining[0].chat_jid, other_chat); +} + +#[tokio::test] +async fn review_245_clear_blocks_late_completion_after_restart() { + review_245_deleted_pending_import(false).await; +} +#[tokio::test] +async fn review_245_delete_blocks_late_completion_after_restart() { + review_245_deleted_pending_import(true).await; +} + +async fn review_245_shared_placeholder(edit: bool) { + 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 operation: wa::WebMessageInfo = historical("OP", None, true, AT - 10); + let mut key: wa::MessageKey = Default::default(); + key.id = Some("MISSING".into()); + key.from_me = Some(true); + let mut protocol: wa::message::ProtocolMessage = Default::default(); + protocol.key = MessageField::some(key); + protocol.r#type = Some(if edit { + wa::message::protocol_message::Type::MESSAGE_EDIT + } else { + wa::message::protocol_message::Type::REVOKE + }); + if edit { + protocol.edited_message = MessageField::some(wa::Message::text("edited history")); + } + let mut message: wa::Message = Default::default(); + message.protocol_message = MessageField::some(protocol); + operation.message = MessageField::some(message); + store + .apply_history_bundle(&pending, &compressed(vec![operation])) + .unwrap(); + store.flush().await.unwrap(); + let rows = store.messages(&jid(GROUP), None, 100).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!( + store.chat(&jid(GROUP)).await.unwrap().unwrap().unread_count, + 0, + "shared amendment created unread attention" + ); + assert!(rows[0].received_as_shared_history); + store.reconcile_chat(&jid(GROUP)).unwrap(); + store.flush().await.unwrap(); + assert_eq!( + store.chat(&jid(GROUP)).await.unwrap().unwrap().unread_count, + 0 + ); +} +#[tokio::test] +async fn review_245_shared_edit_placeholder_stays_read() { + review_245_shared_placeholder(true).await; +} +#[tokio::test] +async fn review_245_shared_revoke_placeholder_stays_read() { + review_245_shared_placeholder(false).await; +} + +#[tokio::test] +async fn review_245_sharer_reaction_can_be_replaced_and_removed_live() { + 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 message = historical("TARGET", Some(OTHER), false, AT - 20); + let mut key: wa::MessageKey = Default::default(); + key.from_me = Some(true); + key.id = Some("REACTION".into()); + key.remote_jid = Some(GROUP.into()); + let mut reaction: wa::Reaction = Default::default(); + reaction.key = MessageField::some(key); + reaction.text = Some("old".into()); + reaction.sender_timestamp_ms = Some((AT - 10) * 1000); + message.reactions.push(reaction); + store + .apply_history_bundle(&pending, &compressed(vec![message])) + .unwrap(); + store.flush().await.unwrap(); + let imported = store.reactions(&jid(GROUP), "TARGET").await.unwrap(); + assert_eq!(imported.len(), 1); + assert_eq!( + imported[0].sender_jid, + jid(PEER), + "sharer's reaction was imported as local" + ); + for (index, emoji) in ["new", ""].into_iter().enumerate() { + let mut key: wa::MessageKey = Default::default(); + key.id = Some("TARGET".into()); + let mut reaction: wa::message::ReactionMessage = Default::default(); + reaction.key = MessageField::some(key); + reaction.text = Some(emoji.into()); + let mut message: wa::Message = Default::default(); + message.reaction_message = MessageField::some(reaction); + feed( + &store, + [message_event( + message, + incoming_info(GROUP, PEER, "LIVE-REACTION", AT + index as i64), + )], + ) + .await; + let rows = store.reactions(&jid(GROUP), "TARGET").await.unwrap(); + if emoji.is_empty() { + assert!(rows.is_empty()); + } else { + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].emoji, emoji); + } + } +} + +#[tokio::test] +async fn review_245_ranged_clear_retires_only_covered_attachment_keys() { + let (db, store) = test_store().await; + own(&db).await; + for (id, at) in [ + ("OLDER", AT - 1), + ("COVERED", AT), + ("TWIN", AT), + ("NEWER", AT + 1), + ] { + feed( + &store, + [message_event( + bundle(&[OWN]), + incoming_info(GROUP, PEER, id, at), + )], + ) + .await; + } + let mut range = range_up_to(AT); + let mut key: wa::MessageKey = Default::default(); + key.id = Some("COVERED".into()); + let mut boundary: wa::sync_action_value::SyncActionMessage = Default::default(); + boundary.key = MessageField::some(key); + range.as_option_mut().unwrap().messages.push(boundary); + let mut action: wa::sync_action_value::ClearChatAction = Default::default(); + action.message_range = range; + feed( + &store, + [Event::ClearChatUpdate( + wacore::types::events::ClearChatUpdate::builder() + .jid(jid(GROUP)) + .delete_starred(false) + .delete_media(false) + .timestamp(ts(AT + 2)) + .action(Box::new(action)) + .from_full_sync(false) + .build(), + )], + ) + .await; + let pending = store.pending_history_bundles().await.unwrap(); + assert_eq!( + pending + .iter() + .map(|b| b.msg_id.as_str()) + .collect::>(), + ["TWIN", "NEWER"] + ); +} + +#[tokio::test] +async fn review_245_shared_amendment_keeps_existing_live_unread_eligibility() { + let (db, store) = test_store().await; + own(&db).await; + receive(&store, wa::Message::text("live"), "TARGET").await; + receive(&store, bundle(&[OWN]), "BUNDLE").await; + let pending = store.pending_history_bundles().await.unwrap().remove(0); + let mut message = historical("EDIT", None, true, AT - 10); + let mut key: wa::MessageKey = Default::default(); + key.id = Some("TARGET".into()); + key.from_me = Some(true); + let mut protocol: wa::message::ProtocolMessage = Default::default(); + protocol.r#type = Some(wa::message::protocol_message::Type::MESSAGE_EDIT); + protocol.key = MessageField::some(key); + protocol.edited_message = MessageField::some(wa::Message::text("history edit")); + let mut content: wa::Message = Default::default(); + content.protocol_message = MessageField::some(protocol); + message.message = MessageField::some(content); + 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.len(), 1); + assert!( + !rows[0].received_as_shared_history, + "a live target was relabeled as read backfill" + ); + assert_eq!( + store.chat(&jid(GROUP)).await.unwrap().unwrap().unread_count, + 1 + ); + store.reconcile_chat(&jid(GROUP)).unwrap(); + store.flush().await.unwrap(); + assert_eq!( + store.chat(&jid(GROUP)).await.unwrap().unwrap().unread_count, + 1 + ); +} diff --git a/docs/shared-group-history.md b/docs/shared-group-history.md index 6ccb7052..a00a5843 100644 --- a/docs/shared-group-history.md +++ b/docs/shared-group-history.md @@ -144,3 +144,41 @@ An earlier combined `cargo test` invocation passed 281 session tests but exhaust 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. + +## Post-merge review corrections + +Review of merged PR #246 found three reproducible gaps. Pending attachments +survived clear/delete-chat, missing-target shared edits and revokes created +unread placeholders, and aggregated reaction keys still described the sharing +account's perspective. Synthetic regressions reproduce each on merged base +`5ee0050c279987abf16c6418d9ff9c1251ce975a`. + +Clear/delete now retires pending attachment envelopes covered by the action's +existing timestamp/key range in the same writer transaction. Retired keys remain +for deduplication. A downloaded body held by an older task cannot consume that +retired envelope after restart, and re-delivery cannot queue it again. Other +chats and envelopes outside the action's range remain untouched. This is local +store deletion behavior, not a claim about server-side history retention. + +Shared edit/revoke insertions retain the read-backfill marker and contribute no +unread delta. Existing live targets keep their own unread eligibility. Aggregated +reaction keys are translated before ordinary history import, allowing the +sharing account's later live replacement/removal to address the same reaction. + +The same verified Web snapshot supports the reaction interpretation. +`WAWebGroupHistorySerializationUtils.serializeSingleMessage` serializes reaction +keys from the original message keys and sets their `unread` field false. Its +identical factory appears in bundle `414c0febcdba83636838a75d6f3f50ef4ad2cb39dbfaa2a16e3a6ece37fc86cb`, +bytes 436848..441175, and bundle `9b681e073f14f9d50e3dc12b456e4f980b554eff4d58827bc91c90a697e9ca38`, +bytes 1266844..1271171. `WAWebAddonProcessMsgsUtils.parseHistorySyncMsg` forwards +aggregates to add-on conversion in bundle `430cb8979511aada95178605faa35a8483446f677ce6d1fe0fd66333c7879eee`, +bytes 343106..347691. The `WAWebE2EProtoUtils` key-perspective conversion is +recorded above. No authenticated reproduction or new server contract is claimed. + +The final follow-up fixture (`tests/shared_history.rs`, SHA-256 +`c8acac7866197ec32a669292e6ef07c06f3ddde57b6452bbf4b6f025c2971aa6`) +was run unchanged on the merged base: six `review_245` checks failed and the +existing-live-target eligibility guard passed. All seven pass with the patch. +Validation also passed formatting, chat-store Clippy with all targets/features +and warnings denied, and the complete chat-store Nextest suite (397 passed, +one skipped). Chat-store doctests completed with three ignored examples.