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
1 change: 1 addition & 0 deletions crates/chat-store/src/store/chat_rows.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 5 additions & 1 deletion crates/chat-store/src/store/edit.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -28,6 +30,7 @@ pub(super) fn apply_edit(
new_kind: &str,
new_proto: &[u8],
ts_ms: i64,
shared: bool,
changes: &mut ChangeSet,
) -> QueryResult<bool> {
use schema::messages::dsl;
Expand Down Expand Up @@ -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)?
Expand All @@ -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)
Expand Down
2 changes: 2 additions & 0 deletions crates/chat-store/src/store/history_sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -456,6 +457,7 @@ pub(super) fn apply_history_message(
target_from_me,
ts_ms,
false,
shared,
cs,
)? {
cs.chats = true;
Expand Down
2 changes: 2 additions & 0 deletions crates/chat-store/src/store/inbound.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -180,6 +181,7 @@ pub(super) fn apply_inbound(
target_from_me,
ts_ms,
false,
false,
cs,
)? {
cs.chats = true;
Expand Down
6 changes: 5 additions & 1 deletion crates/chat-store/src/store/revoke.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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<bool> {
use schema::messages::dsl;
Expand Down Expand Up @@ -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)?
Expand Down Expand Up @@ -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,
)?;
Expand Down
66 changes: 61 additions & 5 deletions crates/chat-store/src/store/shared_history.rs
Original file line number Diff line number Diff line change
Expand Up @@ -154,18 +154,25 @@ 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<usize> {
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::<Integer,_>(device).bind::<Text,_>(&bundle.chat_jid).bind::<Text,_>(&bundle.msg_id).bind::<Text,_>(&bundle.sender_jid).bind::<Binary,_>(&bundle.proto).bind::<BigInt,_>(bundle.timestamp_ms).execute(conn)
}

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::<Integer,_>(device).bind::<Text,_>(&bundle.chat_jid).bind::<Text,_>(&bundle.msg_id).bind::<Text,_>(&bundle.sender_jid).bind::<Binary,_>(&bundle.proto).bind::<BigInt,_>(bundle.timestamp_ms).execute(conn)?;
if changed == 0 {
if complete(conn, device, bundle)? == 0 {
return Ok(());
}
let Some(history) = history else {
Expand Down Expand Up @@ -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::<Jid>().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,
Expand Down Expand Up @@ -437,3 +467,29 @@ pub(super) fn repair_legacy(
.bind::<Text, _>(chat).bind::<BigInt, _>(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::<Integer, _>(device).bind::<Text, _>(chat).load::<PendingHistoryBundle>(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(())
}
2 changes: 2 additions & 0 deletions crates/chat-store/src/store/writer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -496,6 +496,7 @@ fn apply_writer_msg(
kind,
proto,
*timestamp_ms,
false,
cs,
)? {
cs.chats = true;
Expand All @@ -520,6 +521,7 @@ fn apply_writer_msg(
*target_from_me,
*timestamp_ms,
true,
false,
cs,
)? {
cs.chats = true;
Expand Down
Loading
Loading