Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
DROP TABLE shared_group_history;
ALTER TABLE messages DROP COLUMN received_as_shared_history;
Original file line number Diff line number Diff line change
@@ -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);
3 changes: 3 additions & 0 deletions crates/chat-store/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,

Expand Down
1 change: 1 addition & 0 deletions crates/chat-store/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
25 changes: 24 additions & 1 deletion crates/chat-store/src/materialize.rs
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ pub(crate) enum MessageOp {
target_from_me: bool,
target_participant: Option<String>,
},
/// A deliberately shared group-history attachment, persisted before ACK.
HistoryBundle,
/// Protocol/bookkeeping payloads that don't belong in a chat log.
Ignore,
}
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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;
}
Expand All @@ -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<String> {
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))
Expand Down Expand Up @@ -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)]
Expand Down
6 changes: 6 additions & 0 deletions crates/chat-store/src/queries.rs
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,7 @@ pub(crate) struct MessageRow {
edited_at_ms: Option<i64>,
revoked: bool,
local_revoke_placeholder: bool,
received_as_shared_history: bool,
}

/// Fold an author-equivalent legacy copy for a read without hiding a
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -492,6 +495,7 @@ impl From<MessageRow> 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,
}
}
Expand Down Expand Up @@ -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)
Expand Down
1 change: 1 addition & 0 deletions crates/chat-store/src/schema.rs
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ diesel::table! {
edited_at_ms -> Nullable<BigInt>,
revoked -> Bool,
local_revoke_placeholder -> Bool,
received_as_shared_history -> Bool,
}
}

Expand Down
27 changes: 24 additions & 3 deletions crates/chat-store/src/store/history_sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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<PendingQuote>,
shared: bool,
) -> QueryResult<()> {
let Some(key) = wmi.key.as_option() else {
return Ok(());
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -440,7 +461,7 @@ fn apply_history_message(
cs.chats = true;
}
}
MessageOp::Ignore => {}
MessageOp::HistoryBundle | MessageOp::Ignore => {}
}
}

Expand Down
29 changes: 27 additions & 2 deletions crates/chat-store/src/store/inbound.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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(())
Expand Down
13 changes: 9 additions & 4 deletions crates/chat-store/src/store/message_identity.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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::<Integer, _>(device_id)
Expand All @@ -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
Expand All @@ -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 > ?
) <= ?",
)
Expand Down
31 changes: 31 additions & 0 deletions crates/chat-store/src/store/message_rows.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading
Loading