diff --git a/crates/audio/src/lib.rs b/crates/audio/src/lib.rs index 1f546afc..ab672626 100644 --- a/crates/audio/src/lib.rs +++ b/crates/audio/src/lib.rs @@ -29,6 +29,7 @@ mod encoder; /// Opus packets, in the OGG stream WhatsApp expects. The container is not the /// codec, and only the codec was ever the problem here. mod ogg_opus; +mod playback_start; mod player; mod recorder; mod resample; diff --git a/crates/audio/src/playback_start.rs b/crates/audio/src/playback_start.rs new file mode 100644 index 00000000..3ccc3511 --- /dev/null +++ b/crates/audio/src/playback_start.rs @@ -0,0 +1,61 @@ +//! Evidence that a sink advanced, independent of seeking or decoder completion. +use std::sync::atomic::{AtomicBool, Ordering}; + +#[derive(Default)] +pub(crate) struct PlaybackStart { + observed: AtomicBool, + reported: AtomicBool, +} + +impl PlaybackStart { + #[cfg(any(not(target_family = "wasm"), test))] + pub(crate) fn samples(&self, consumed: usize) { + if consumed > 0 { + self.observed.store(true, Ordering::Release); + } + } + + #[cfg(any(target_family = "wasm", test))] + pub(crate) fn clock(&self, playing: bool, now: f64, started_at: f64) { + if playing && now > started_at { + self.observed.store(true, Ordering::Release); + } + } + + pub(crate) fn take(&self, accepted: bool) -> bool { + accepted + && self.observed.load(Ordering::Acquire) + && !self.reported.swap(true, Ordering::AcqRel) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn native_requires_consumed_samples_and_an_accepted_stream() { + let start = PlaybackStart::default(); + assert!(!start.take(true)); // load or successful stream installation alone + start.samples(0); + assert!(!start.take(true)); // empty callback or seek to the end + start.samples(24); + assert!(!start.take(false)); // output start failed + assert!(start.take(true)); + start.samples(48); + assert!(!start.take(true)); // repeated callbacks and polling + } + + #[test] + fn web_requires_running_clock_and_reports_short_clips_after_end() { + let start = PlaybackStart::default(); + start.clock(true, 5.0, 5.0); // suspended context, including seek offset + start.clock(false, 6.0, 5.0); // paused or failed node + assert!(!start.take(true)); + start.clock(true, 6.001, 6.0); // actual playing or natural ended callback + assert!(start.take(true)); + start.clock(true, 8.0, 7.0); // resume or seek replaces the node + assert!(!start.take(true)); + assert!(!PlaybackStart::default().take(true)); // stopped/new clip + } +} diff --git a/crates/audio/src/player.rs b/crates/audio/src/player.rs index 1a58891e..84648208 100644 --- a/crates/audio/src/player.rs +++ b/crates/audio/src/player.rs @@ -23,6 +23,7 @@ mod cpal_output { use tokio::sync::oneshot; use super::PlayerError; + use crate::playback_start::PlaybackStart; /// Samples ready for the output callback. /// @@ -39,6 +40,7 @@ mod cpal_output { /// Audio player for PTT voice messages. pub struct AudioPlayer { stream: Option, + started: Arc, is_playing: Arc, position: Arc, total_samples: u64, @@ -84,6 +86,7 @@ mod cpal_output { pub fn new() -> Self { Self { stream: None, + started: Arc::new(PlaybackStart::default()), is_playing: Arc::new(AtomicBool::new(false)), position: Arc::new(AtomicUsize::new(0)), finished: Arc::new(AtomicBool::new(false)), @@ -119,6 +122,11 @@ mod cpal_output { rx } + /// True once per loaded clip, after the output callback consumed samples. + pub fn take_started(&mut self) -> bool { + self.started.take(self.stream.is_some()) + } + pub fn is_playing(&self) -> bool { self.is_playing.load(Ordering::Relaxed) } @@ -329,6 +337,7 @@ mod cpal_output { config.sample_rate, output_channels, sample_format ); + self.started = Arc::new(PlaybackStart::default()); let is_playing = self.is_playing.clone(); let position = self.position.clone(); position.store(0, Ordering::Relaxed); @@ -351,6 +360,7 @@ mod cpal_output { is_playing, finished, completion_tx, + self.started.clone(), ), SampleFormat::I16 => build_stream::( &device, @@ -360,6 +370,7 @@ mod cpal_output { is_playing, finished, completion_tx, + self.started.clone(), ), SampleFormat::U16 => build_stream::( &device, @@ -369,6 +380,7 @@ mod cpal_output { is_playing, finished, completion_tx, + self.started.clone(), ), other => Err(PlayerError::StreamError(format!( "unsupported output sample format {other:?}" @@ -400,6 +412,7 @@ mod cpal_output { pub fn stop(&mut self) { self.stream.take(); + self.started = Arc::new(PlaybackStart::default()); self.is_playing.store(false, Ordering::Relaxed); self.finished.store(false, Ordering::Relaxed); self.position.store(0, Ordering::Relaxed); @@ -434,6 +447,7 @@ mod cpal_output { /// Build the output stream for the device's sample format, converting our /// f32 samples in the callback (same dispatch as call_device's speaker path). + #[allow(clippy::too_many_arguments)] fn build_stream>( device: &cpal::Device, config: StreamConfig, @@ -442,6 +456,7 @@ mod cpal_output { is_playing: Arc, finished: Arc, completion_tx: Arc>>>, + started: Arc, ) -> Result { let err_is_playing = is_playing.clone(); let err_finished = finished.clone(); @@ -451,10 +466,14 @@ mod cpal_output { config, move |data: &mut [T], _: &cpal::OutputCallbackInfo| { let mut pos = position.load(Ordering::Relaxed); + let before = pos; for sample in data.iter_mut() { let s = if pos < audio.len() { let s = audio[pos]; + if pos == before { + started.samples(1); + } pos += 1; s } else { diff --git a/crates/audio/src/web/player.rs b/crates/audio/src/web/player.rs index c8a73946..410ed032 100644 --- a/crates/audio/src/web/player.rs +++ b/crates/audio/src/web/player.rs @@ -24,6 +24,7 @@ use wasm_bindgen::prelude::Closure; use wasm_bindgen_futures::{JsFuture, spawn_local}; use web_sys::{AudioBuffer, AudioBufferSourceNode, AudioContext}; +use crate::playback_start::PlaybackStart; use crate::player::PlayerError; /// Encoded voice note waiting for the browser's asynchronous decoder. @@ -93,6 +94,7 @@ struct Playing { duration: f64, is_playing: bool, finished: bool, + playback_start: PlaybackStart, /// Where a seek asked to be while the clip was still decoding, as a /// fraction — the duration is not known until the buffer is. /// @@ -137,6 +139,7 @@ impl Default for Playing { duration: 0.0, is_playing: false, finished: false, + playback_start: PlaybackStart::default(), pending_seek: None, pending_pause: false, decoding: false, @@ -186,6 +189,7 @@ impl AudioPlayer { /// between the two, and it is the browser's own control rather than a /// second implementation of one. pub fn set_speed(&mut self, speed: f32) { + self.observe_started(); // Bank what has played at the old rate before the new one applies. // `offset` is measured in the clip's own seconds, so without this the // time already played would be re-scaled by a rate it was not played @@ -319,6 +323,25 @@ impl AudioPlayer { (elapsed.clamp(0.0, state.duration)) as f32 } + fn observe_started(&self) { + let now = self + .context + .as_ref() + .map_or(0.0, AudioContext::current_time); + let state = self.state.borrow(); + state.playback_start.clock( + state.is_playing && state.offset < state.duration, + now, + state.started_at, + ); + } + + /// Report playback clock advancement once, independently of seek offsets. + pub fn take_started(&mut self) -> bool { + self.observe_started(); + self.state.borrow().playback_start.take(true) + } + /// How long the clip is, in its own seconds. /// /// Not divided by the speed. `elapsed_secs` already reports a position on @@ -340,6 +363,7 @@ impl AudioPlayer { /// open. Refused once the clip has finished, for the same reason the /// native player refuses it — the caller starts it over instead. pub fn seek(&self, fraction: f32) -> bool { + self.observe_started(); if self.state.borrow().finished { return false; } @@ -550,6 +574,7 @@ impl AudioPlayer { state.offset = 0.0; state.duration = 0.0; state.finished = false; + state.playback_start = PlaybackStart::default(); state.pending_seek = None; state.pending_pause = false; // Back to the default, because both entry points come through here @@ -561,6 +586,7 @@ impl AudioPlayer { /// Stop making sound, keeping the position. pub fn pause(&mut self) { + self.observe_started(); if !self.state.borrow().is_playing { // Same as `seek`: a pause asked for while the clip is decoding // has to be remembered, or the note starts playing on its own the @@ -700,12 +726,18 @@ fn start( let ended = { let state = Rc::clone(state); + let context = context.clone(); Closure::::new(move || { let completion = { let mut state = state.borrow_mut(); if state.generation != generation || !state.is_playing { return; } + state.playback_start.clock( + state.offset < state.duration, + context.current_time(), + state.started_at, + ); state.is_playing = false; state.finished = true; state.offset = state.duration; diff --git a/crates/chat-store/migrations/2026-10-26-140000_audio_played/down.sql b/crates/chat-store/migrations/2026-10-26-140000_audio_played/down.sql new file mode 100644 index 00000000..74311433 --- /dev/null +++ b/crates/chat-store/migrations/2026-10-26-140000_audio_played/down.sql @@ -0,0 +1 @@ +DROP TABLE audio_played_receipts; diff --git a/crates/chat-store/migrations/2026-10-26-140000_audio_played/up.sql b/crates/chat-store/migrations/2026-10-26-140000_audio_played/up.sql new file mode 100644 index 00000000..ddf506ab --- /dev/null +++ b/crates/chat-store/migrations/2026-10-26-140000_audio_played/up.sql @@ -0,0 +1,5 @@ +CREATE TABLE audio_played_receipts ( + device_id INTEGER NOT NULL REFERENCES device(id) ON DELETE CASCADE, + message_row_id BIGINT NOT NULL REFERENCES messages(id) ON DELETE CASCADE, + PRIMARY KEY (device_id, message_row_id) +); diff --git a/crates/chat-store/migrations/2026-10-26-140001_pending_audio_plays/down.sql b/crates/chat-store/migrations/2026-10-26-140001_pending_audio_plays/down.sql new file mode 100644 index 00000000..c617a8bc --- /dev/null +++ b/crates/chat-store/migrations/2026-10-26-140001_pending_audio_plays/down.sql @@ -0,0 +1 @@ +DROP TABLE audio_played_pending; diff --git a/crates/chat-store/migrations/2026-10-26-140001_pending_audio_plays/up.sql b/crates/chat-store/migrations/2026-10-26-140001_pending_audio_plays/up.sql new file mode 100644 index 00000000..1859bb61 --- /dev/null +++ b/crates/chat-store/migrations/2026-10-26-140001_pending_audio_plays/up.sql @@ -0,0 +1,9 @@ +CREATE TABLE audio_played_pending ( + device_id INTEGER NOT NULL REFERENCES device(id) ON DELETE CASCADE, + chat_jid TEXT NOT NULL, + msg_id TEXT NOT NULL, + author_jid TEXT NOT NULL, + received_at_ms BIGINT NOT NULL, + ambiguous BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (device_id, chat_jid, msg_id, author_jid) +); diff --git a/crates/chat-store/src/queries.rs b/crates/chat-store/src/queries.rs index fc31961e..be695d85 100644 --- a/crates/chat-store/src/queries.rs +++ b/crates/chat-store/src/queries.rs @@ -1647,6 +1647,50 @@ impl ChatStore { Ok(messages.into_iter().next()) } + /// Whether this account already submitted the played receipt or learned it + /// from a linked device. Local-only playback does not set this bit. + pub async fn audio_played_receipt_sent(&self, message: &StoredMessage) -> Result { + if message.from_me { + return Ok(false); + } + let device_id = self.device_id(); + let chat = message.chat_jid.to_non_ad_string(); + let sender = message.sender_jid.to_non_ad_string(); + let id = message.id.clone(); + self.db() + .read(move |conn| { + #[derive(diesel::QueryableByName)] + struct Count { + #[diesel(sql_type = diesel::sql_types::BigInt)] + count: i64, + } + let aliases = + crate::lid::chat_key_candidates(conn, device_id, &chat).map_err(db_err)?; + for chat in aliases { + let rows = crate::store::message_identity::matching_rows( + conn, device_id, &chat, &id, false, &sender, + ) + .map_err(db_err)?; + for row in rows { + let count = diesel::sql_query( + "SELECT count(*) AS count FROM audio_played_receipts \ + WHERE device_id = ? AND message_row_id = ?", + ) + .bind::(device_id) + .bind::(row.id) + .get_result::(conn) + .map_err(db_err)?; + if count.count > 0 { + return Ok(true); + } + } + } + Ok(false) + }) + .await + .map_err(Into::into) + } + pub async fn message(&self, chat: &Jid, msg_id: &str) -> Result> { self.message_by_ownership(chat, msg_id, None).await } diff --git a/crates/chat-store/src/schema.rs b/crates/chat-store/src/schema.rs index ebab96d0..61e25bef 100644 --- a/crates/chat-store/src/schema.rs +++ b/crates/chat-store/src/schema.rs @@ -162,3 +162,14 @@ diesel::allow_tables_to_appear_in_same_query!( message_identity_repair_pending, contact_name_removals ); + +diesel::table! { + audio_played_pending (device_id, chat_jid, msg_id, author_jid) { + device_id -> Integer, + chat_jid -> Text, + msg_id -> Text, + author_jid -> Text, + received_at_ms -> BigInt, + ambiguous -> Bool, + } +} diff --git a/crates/chat-store/src/store/audio_played.rs b/crates/chat-store/src/store/audio_played.rs new file mode 100644 index 00000000..20e65fec --- /dev/null +++ b/crates/chat-store/src/store/audio_played.rs @@ -0,0 +1,155 @@ +//! Linked playback can arrive before its message. Keep bounded identity evidence, +//! then resolve it after the whole writer batch, never as an outgoing job. +use diesel::prelude::*; +use diesel::sql_types::{BigInt, Integer}; +use wacore_binary::{Jid, JidExt as _}; + +use super::message_identity::{authors_match, merge_rows}; +use super::writer::ChangeSet; +use crate::schema::{audio_played_pending as pending, messages}; + +const MAX_PENDING_PER_ACCOUNT: i64 = 4096; + +pub(super) fn has_pending(conn: &mut SqliteConnection, device: i32) -> QueryResult { + diesel::select(diesel::dsl::exists( + pending::table.filter(pending::device_id.eq(device)), + )) + .get_result(conn) +} + +pub(super) fn observe( + conn: &mut SqliteConnection, + device: i32, + chat: &str, + ids: &[String], + author: &str, + received_at_ms: i64, + changes: &mut ChangeSet, +) -> QueryResult<()> { + for id in ids { + diesel::insert_into(pending::table) + .values(( + pending::device_id.eq(device), + pending::chat_jid.eq(chat), + pending::msg_id.eq(id), + pending::author_jid.eq(author), + pending::received_at_ms.eq(received_at_ms), + )) + .on_conflict_do_nothing() + .execute(conn)?; + } + let evicted = diesel::sql_query( + "DELETE FROM audio_played_pending WHERE rowid IN \ + (SELECT rowid FROM audio_played_pending WHERE device_id = ? \ + ORDER BY received_at_ms DESC, rowid DESC LIMIT -1 OFFSET ?)", + ) + .bind::(device) + .bind::(MAX_PENDING_PER_ACCOUNT) + .execute(conn)?; + if evicted > 0 { + log::warn!( + "chat-store: evicted {evicted} unmatched playback observations at the account limit" + ); + } + changes.pending_audio_chats.insert(chat.to_string()); + Ok(()) +} + +pub(super) fn apply( + conn: &mut SqliteConnection, + device: i32, + chat: &str, + changes: &mut ChangeSet, +) -> QueryResult<()> { + let aliases = crate::lid::chat_key_candidates(conn, device, chat)?; + let observations: Vec<(String, String, String, bool)> = pending::table + .filter( + pending::device_id + .eq(device) + .and(pending::chat_jid.eq_any(&aliases)), + ) + .select(( + pending::chat_jid, + pending::msg_id, + pending::author_jid, + pending::ambiguous, + )) + .load(conn)?; + for (observed_chat, id, author, ambiguous) in observations { + if ambiguous { + continue; + } + if author.is_empty() + && observed_chat.parse::().map_or(true, |jid| { + jid.is_group() || jid.is_status_broadcast() || jid.is_broadcast_list() + }) + { + // These message keys include an author. A first arriving row cannot prove it. + continue; + } + let key = pending::table.filter( + pending::device_id + .eq(device) + .and(pending::chat_jid.eq(&observed_chat)) + .and(pending::msg_id.eq(&id)) + .and(pending::author_jid.eq(&author)), + ); + let candidates: Vec<(i64, String, String)> = messages::table + .filter( + messages::device_id + .eq(device) + .and(messages::chat_jid.eq_any(&aliases)) + .and(messages::msg_id.eq(&id)) + .and(messages::from_me.eq(false)) + .and(messages::revoked.eq(false)), + ) + .select((messages::id, messages::chat_jid, messages::sender_jid)) + .load(conn)?; + let mut rows = Vec::new(); + for row in candidates { + if author.is_empty() || authors_match(conn, device, false, &author, false, &row.2)? { + rows.push(row); + } + } + let Some(first) = rows.first() else { + continue; + }; + let mut same_author = true; + for row in rows.iter().skip(1) { + same_author &= authors_match(conn, device, false, &first.2, false, &row.2)?; + } + if !same_author { + // Unknown group author cannot be recovered by deleting a competing row later. + diesel::update(key) + .set(pending::ambiguous.eq(true)) + .execute(conn)?; + continue; + } + let (row_id, row_chat) = if rows.len() > 1 { + let ids: Vec<_> = rows.iter().map(|row| row.0).collect(); + let survivor = merge_rows(conn, device, &id, chat, &ids, false)?; + changes + .message_chats + .extend(rows.iter().map(|row| row.1.clone())); + (survivor.id, chat.to_string()) + } else { + (first.0, first.1.clone()) + }; + let changed = diesel::update( + messages::table.filter(messages::id.eq(row_id).and(messages::status.lt(5))), + ) + .set(messages::status.eq(5)) + .execute(conn)?; + diesel::sql_query( + "INSERT OR IGNORE INTO audio_played_receipts (device_id, message_row_id) VALUES (?, ?)", + ) + .bind::(device) + .bind::(row_id) + .execute(conn)?; + diesel::delete(key).execute(conn)?; + if changed > 0 { + changes.message_chats.insert(row_chat); + } + } + Ok(()) +} diff --git a/crates/chat-store/src/store/message_identity.rs b/crates/chat-store/src/store/message_identity.rs index d58cb1a1..57ba363b 100644 --- a/crates/chat-store/src/store/message_identity.rs +++ b/crates/chat-store/src/store/message_identity.rs @@ -15,7 +15,7 @@ use crate::types::MessageStatus; #[derive(Debug, Clone)] pub(crate) struct MessageOwner { - pub(super) id: i64, + pub(crate) id: i64, pub(super) sender_jid: String, pub(super) from_me: bool, } @@ -66,7 +66,7 @@ pub(crate) fn authors_match( } /// Message rows with this stanza id which belong to the supplied author. -pub(super) fn matching_rows( +pub(crate) fn matching_rows( conn: &mut SqliteConnection, device_id: i32, chat: &str, @@ -228,6 +228,13 @@ pub(crate) fn merge_rows( .filter_map(|row| (row.id != survivor_id).then_some(row.id)) .collect(); if !duplicates.is_empty() { + // Submission follows the proven message identity when its duplicate row is removed. + for duplicate in &duplicates { + diesel::sql_query("INSERT OR IGNORE INTO audio_played_receipts (device_id, message_row_id) SELECT device_id, ? FROM audio_played_receipts WHERE device_id = ? AND message_row_id = ?") + .bind::(survivor_id) + .bind::(device_id) + .bind::(*duplicate).execute(conn)?; + } diesel::delete(dsl::messages.filter(dsl::id.eq_any(duplicates))).execute(conn)?; } diesel::update(dsl::messages.filter(dsl::id.eq(survivor_id))) diff --git a/crates/chat-store/src/store/mod.rs b/crates/chat-store/src/store/mod.rs index b4c8991d..cba87028 100644 --- a/crates/chat-store/src/store/mod.rs +++ b/crates/chat-store/src/store/mod.rs @@ -8,6 +8,7 @@ //! does per event, split by the kind of event each one materializes. mod ack; +mod audio_played; mod avatar; mod chat_names; mod chat_rows; @@ -182,6 +183,12 @@ pub(crate) enum WriterMsg { chat: Jid, msg_id: String, }, + AudioPlayed { + chat: String, + message_id: String, + sender: String, + receipt_sent: bool, + }, StatusWatched { chat: Jid, msg_ids: Vec, @@ -768,6 +775,23 @@ impl ChatStore { .map_err(|error| ChatStoreError::Store(StoreError::Validation(error.to_string()))) } + /// Record local playback separately from receipt transport submission. + /// Use `flush` before treating either write as durable. + pub fn record_audio_played( + &self, + message: &crate::StoredMessage, + receipt_sent: bool, + ) -> Result<()> { + self.tx + .send(WriterMsg::AudioPlayed { + chat: message.chat_jid.to_non_ad_string(), + message_id: message.id.clone(), + sender: message.sender_jid.to_non_ad_string(), + receipt_sent, + }) + .map_err(|e| ChatStoreError::Store(StoreError::Validation(e.to_string()))) + } + /// Record that these status updates have been watched on this device. /// /// The same place WhatsApp Web keeps it — the message's own ack moved to @@ -1256,11 +1280,14 @@ mod migration_tests { .expect("an empty retirement journal downgrade is reversible"); assert!(!has_table(&store, "account_retirement").await); assert!(has_table(&store, "session_recovery").await); + // The version-targeted retirement rewind already removed newer metadata. + assert!(!has_table(&store, "audio_played_pending").await); + assert!(!has_table(&store, "audio_played_receipts").await); store .shared() .run(revert_last_chat_migration) .await - .expect("session-recovery downgrade is reversible"); + .expect("session recovery downgrade is reversible"); assert!(!has_table(&store, "session_recovery").await); assert!(has_table(&store, "shared_group_history").await); store diff --git a/crates/chat-store/src/store/receipt.rs b/crates/chat-store/src/store/receipt.rs index 6af447c1..69a69c7b 100644 --- a/crates/chat-store/src/store/receipt.rs +++ b/crates/chat-store/src/store/receipt.rs @@ -40,6 +40,27 @@ pub(super) fn apply_receipt( // be keyed by either identity (or split) — route to where it lives // so the read state lands on the real rows, not a stray twin. let chat = route_chat(conn, device_id, chat, cs)?; + if receipt.r#type == ReceiptType::PlayedSelf { + let author = if receipt.source.chat.is_pn() || receipt.source.chat.is_lid() { + chat.clone() + } else { + receipt + .source + .recipient + .as_ref() + .map(|jid| jid.to_non_ad_string()) + .unwrap_or_default() + }; + super::audio_played::observe( + conn, + device_id, + &chat, + &message_ids, + &author, + ts_ms, + cs, + )?; + } // Read on another of our devices — up to the covered messages. // WA read state is "read up to X": the boundary is the newest // covered row (falling back to the receipt's own timestamp). diff --git a/crates/chat-store/src/store/writer.rs b/crates/chat-store/src/store/writer.rs index 41400293..8a2a29c1 100644 --- a/crates/chat-store/src/store/writer.rs +++ b/crates/chat-store/src/store/writer.rs @@ -42,6 +42,7 @@ pub(crate) struct ChangeSet { pub(crate) chats: bool, pub(crate) contacts: bool, pub(crate) message_chats: BTreeSet, + pub(crate) pending_audio_chats: BTreeSet, } pub(super) async fn writer_loop( @@ -251,6 +252,19 @@ async fn attempt_batch( return Err(error); } } + if !cs.pending_audio_chats.is_empty() + || (!cs.message_chats.is_empty() + && super::audio_played::has_pending(conn, device_id)?) + { + let played_chats: Vec<_> = cs + .message_chats + .union(&cs.pending_audio_chats) + .cloned() + .collect(); + for chat in played_chats { + super::audio_played::apply(conn, device_id, &chat, &mut cs)?; + } + } Ok(cs) }); result @@ -575,6 +589,33 @@ fn apply_writer_msg( WriterMsg::GroupHierarchies(writes) => { crate::store::group_hierarchy::apply_group_hierarchies(conn, device_id, writes, cs) } + WriterMsg::AudioPlayed { + chat, + message_id, + sender, + receipt_sent, + } => { + use schema::messages::dsl; + let chat = route_chat(conn, device_id, chat.clone(), cs)?; + let rows = super::message_identity::matching_rows( + conn, device_id, &chat, message_id, false, sender, + )?; + for row in rows { + let changed = + diesel::update(dsl::messages.filter(dsl::id.eq(row.id).and(dsl::status.lt(5)))) + .set(dsl::status.eq(5)) + .execute(conn)?; + if *receipt_sent { + diesel::sql_query("INSERT OR IGNORE INTO audio_played_receipts (device_id, message_row_id) VALUES (?, ?)") + .bind::(device_id) + .bind::(row.id).execute(conn)?; + } + if changed > 0 { + cs.message_chats.insert(chat.clone()); + } + } + Ok(()) + } WriterMsg::StatusWatched { chat, msg_ids } => { // Routed like every other write that targets a row. The broadcast // this is called with today routes to itself, but the method is diff --git a/crates/chat-store/tests/receipts.rs b/crates/chat-store/tests/receipts.rs index 2eddf0e0..a481ae75 100644 --- a/crates/chat-store/tests/receipts.rs +++ b/crates/chat-store/tests/receipts.rs @@ -925,3 +925,537 @@ async fn migration_folds_device_suffixed_rows() { [("111000011112222@lid", 4), ("222000011112222@lid", 4)] ); } + +#[tokio::test] +async fn self_play_is_distinct_from_read_and_survives_duplicate_and_stale_reads() { + let (_db, store) = test_store().await; + feed( + &store, + [message_event( + wa::Message::text("voice fixture"), + incoming_info(PEER, PEER, "VOICE", 100), + )], + ) + .await; + feed( + &store, + [receipt( + jid(PEER), + jid(PEER), + &["VOICE"], + ReceiptType::ReadSelf, + ts(101), + )], + ) + .await; + assert_ne!( + store + .message(&jid(PEER), "VOICE") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); + for ty in [ + ReceiptType::PlayedSelf, + ReceiptType::PlayedSelf, + ReceiptType::ReadSelf, + ] { + feed( + &store, + [receipt(jid(PEER), jid(PEER), &["VOICE"], ty, ts(101))], + ) + .await; + } + assert_eq!( + store + .message(&jid(PEER), "VOICE") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); +} + +#[tokio::test] +async fn local_audio_playback_and_transport_submission_have_separate_durable_state() { + let (_db, store) = test_store().await; + feed( + &store, + [message_event( + wa::Message::text("voice fixture"), + incoming_info(PEER, PEER, "LOCAL-PLAY", 100), + )], + ) + .await; + let row = store + .message(&jid(PEER), "LOCAL-PLAY") + .await + .unwrap() + .unwrap(); + // Offline playback or a transport failure records the observation but never a send. + store.record_audio_played(&row, false).unwrap(); + store.flush().await.unwrap(); + assert_eq!( + store + .message(&jid(PEER), "LOCAL-PLAY") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); + assert!(!store.audio_played_receipt_sent(&row).await.unwrap()); + store.record_audio_played(&row, true).unwrap(); + store.record_audio_played(&row, false).unwrap(); + store.flush().await.unwrap(); + assert!(store.audio_played_receipt_sent(&row).await.unwrap()); + // Reopen the projection over the same synthetic database. + let reopened = ChatStore::new(&_db).await.unwrap(); + assert!(reopened.audio_played_receipt_sent(&row).await.unwrap()); + assert_eq!( + reopened + .message(&jid(PEER), "LOCAL-PLAY") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); +} + +#[tokio::test] +async fn audio_playback_writes_cannot_cross_accounts_or_mark_own_audio() { + let (db, store) = test_store().await; + let db_b = db.database().store(2); + db.database().provision_device(2).await.unwrap(); + let other = ChatStore::new(&db_b).await.unwrap(); + feed( + &store, + [message_event( + wa::Message::text("synthetic"), + incoming_info(PEER, PEER, "ISOLATED", 100), + )], + ) + .await; + let incoming = store + .message(&jid(PEER), "ISOLATED") + .await + .unwrap() + .unwrap(); + let mut stale_identity = incoming.clone(); + stale_identity.id = "REPLACED-ROW".into(); + store.record_audio_played(&stale_identity, true).unwrap(); + store.flush().await.unwrap(); + assert!(!store.audio_played_receipt_sent(&incoming).await.unwrap()); + other.record_audio_played(&incoming, true).unwrap(); + other.flush().await.unwrap(); + assert!(!other.audio_played_receipt_sent(&incoming).await.unwrap()); + assert_ne!( + store + .message(&jid(PEER), "ISOLATED") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); + store.record_audio_played(&incoming, true).unwrap(); + store.flush().await.unwrap(); + assert!(!other.audio_played_receipt_sent(&incoming).await.unwrap()); + store + .record_outgoing( + &jid(PEER), + "OWN-AUDIO", + &wa::Message::text("synthetic own"), + ts(102), + ) + .unwrap(); + store.flush().await.unwrap(); + let own = store + .own_message(&jid(PEER), "OWN-AUDIO") + .await + .unwrap() + .unwrap(); + store.record_audio_played(&own, true).unwrap(); + store.flush().await.unwrap(); + assert!(!store.audio_played_receipt_sent(&own).await.unwrap()); + assert_ne!( + store + .own_message(&jid(PEER), "OWN-AUDIO") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); +} + +#[tokio::test] +async fn linked_play_refuses_an_ambiguous_group_id() { + let (_db, store) = test_store().await; + let authors = ["559900000001@s.whatsapp.net", "111000011112222@lid"]; + feed( + &store, + authors.map(|author| { + message_event( + wa::Message::text("synthetic"), + incoming_info(GROUP, author, "COLLIDING-PLAY", 100), + ) + }), + ) + .await; + feed( + &store, + [receipt( + jid(GROUP), + jid("559900000003@s.whatsapp.net"), + &["COLLIDING-PLAY"], + ReceiptType::PlayedSelf, + ts(101), + )], + ) + .await; + let rows = store.messages(&jid(GROUP), None, 10).await.unwrap(); + assert_eq!(rows.len(), 2); + assert!(rows.iter().all(|row| row.status != MessageStatus::Played)); +} + +#[tokio::test] +async fn played_submission_survives_pn_lid_duplicate_row_fusion() { + let (db, store) = test_store().await; + feed( + &store, + [ + message_event( + wa::Message::text("synthetic pn"), + incoming_info(PEER, PEER, "MERGED-PLAY", 100), + ), + message_event( + wa::Message::text("synthetic lid"), + incoming_info(PEER_LID, PEER_LID, "MERGED-PLAY", 100), + ), + ], + ) + .await; + let pn = store + .message(&jid(PEER), "MERGED-PLAY") + .await + .unwrap() + .unwrap(); + let lid = store + .message(&jid(PEER_LID), "MERGED-PLAY") + .await + .unwrap() + .unwrap(); + assert_ne!(pn.seq, lid.seq); + store.record_audio_played(&lid, true).unwrap(); + store.flush().await.unwrap(); + add_lid_mapping(&db).await; + store.reconcile_chat(&jid(PEER)).unwrap(); + store.flush().await.unwrap(); + let merged = store + .message(&jid(PEER), "MERGED-PLAY") + .await + .unwrap() + .unwrap(); + assert_eq!(merged.status, MessageStatus::Played); + assert!(store.audio_played_receipt_sent(&merged).await.unwrap()); + assert!(store.audio_played_receipt_sent(&pn).await.unwrap()); + assert!(store.audio_played_receipt_sent(&lid).await.unwrap()); + // An observation captured before fusion still targets the proven message. + store.record_audio_played(&lid, true).unwrap(); + store.flush().await.unwrap(); + assert!(store.audio_played_receipt_sent(&merged).await.unwrap()); +} + +fn synthetic_voice() -> wa::Message { + let mut audio = wa::message::AudioMessage::default(); + audio.ptt = Some(true); + let mut message = wa::Message::default(); + message.audio_message = MessageField::some(audio); + message +} + +#[tokio::test] +async fn early_linked_play_survives_reopen_and_pn_lid_materialization() { + let (db, store) = test_store().await; + feed( + &store, + [receipt( + jid(PEER), + jid(PEER), + &["EARLY-PLAY"], + ReceiptType::PlayedSelf, + ts(101), + )], + ) + .await; + add_lid_mapping(&db).await; + let reopened = ChatStore::new(&db).await.unwrap(); + feed( + &reopened, + [message_event( + synthetic_voice(), + incoming_info(PEER_LID, PEER_LID, "EARLY-PLAY", 100), + )], + ) + .await; + let message = reopened + .message(&jid(PEER), "EARLY-PLAY") + .await + .unwrap() + .unwrap(); + assert_eq!(message.status, MessageStatus::Played); + assert!(reopened.audio_played_receipt_sent(&message).await.unwrap()); + feed( + &reopened, + [receipt( + jid(PEER_LID), + jid(PEER_LID), + &["EARLY-PLAY"], + ReceiptType::ReadSelf, + ts(102), + )], + ) + .await; + assert_eq!( + reopened + .message(&jid(PEER), "EARLY-PLAY") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); +} + +#[tokio::test] +async fn early_linked_play_applies_to_a_later_history_message() { + let (_db, store) = test_store().await; + feed( + &store, + [receipt( + jid(PEER), + jid(PEER), + &["HISTORY-PLAY"], + ReceiptType::PlayedSelf, + ts(101), + )], + ) + .await; + let mut key = wa::MessageKey::default(); + key.remote_jid = Some(PEER.into()); + key.id = Some("HISTORY-PLAY".into()); + key.from_me = Some(false); + let mut info = wa::WebMessageInfo::default(); + info.key = MessageField::some(key); + info.message = MessageField::some(synthetic_voice()); + info.message_timestamp = Some(100); + let mut entry = wa::HistorySyncMsg::default(); + entry.message = MessageField::some(info); + let mut conversation = wa::Conversation::default(); + conversation.id = PEER.into(); + conversation.messages = vec![entry]; + let mut history = wa::HistorySync::default(); + history.sync_type = wa::history_sync::HistorySyncType::RECENT; + history.conversations = vec![conversation]; + feed(&store, [history_sync_event(history)]).await; + let message = store + .message(&jid(PEER), "HISTORY-PLAY") + .await + .unwrap() + .unwrap(); + assert_eq!(message.status, MessageStatus::Played); + assert!(store.audio_played_receipt_sent(&message).await.unwrap()); +} + +#[tokio::test] +async fn early_group_play_rejects_ambiguous_authors_in_the_materialization_batch() { + let (_db, store) = test_store().await; + feed( + &store, + [receipt( + jid(GROUP), + jid(PEER), + &["EARLY-COLLISION"], + ReceiptType::PlayedSelf, + ts(101), + )], + ) + .await; + // Commit both authors atomically, like a reconnect/history batch. + let batch: Vec<_> = [PEER, "559900000002@s.whatsapp.net"] + .into_iter() + .map(|author| { + InboundMessage::builder() + .message(Arc::new(synthetic_voice())) + .info(Arc::new(incoming_info( + GROUP, + author, + "EARLY-COLLISION", + 100, + ))) + .build() + }) + .collect(); + store.commit_inbound_batch(&batch).await.unwrap(); + let rows = store.messages(&jid(GROUP), None, 10).await.unwrap(); + assert_eq!(rows.len(), 2); + assert!(rows.iter().all(|row| row.status != MessageStatus::Played)); + for row in rows { + assert!(!store.audio_played_receipt_sent(&row).await.unwrap()); + } +} + +#[tokio::test] +async fn known_group_author_is_preserved_while_waiting_for_its_message() { + let (_db, store) = test_store().await; + let target = "559900000002@s.whatsapp.net"; + let Event::Receipt(mut observed) = receipt( + jid(GROUP), + jid(PEER), + &["AUTHOR-PLAY"], + ReceiptType::PlayedSelf, + ts(101), + ) else { + unreachable!() + }; + observed.source.recipient = Some(jid(target)); + feed(&store, [Event::Receipt(observed)]).await; + feed( + &store, + [message_event( + synthetic_voice(), + incoming_info(GROUP, PEER, "AUTHOR-PLAY", 100), + )], + ) + .await; + assert_ne!( + store + .message(&jid(GROUP), "AUTHOR-PLAY") + .await + .unwrap() + .unwrap() + .status, + MessageStatus::Played + ); + feed( + &store, + [message_event( + synthetic_voice(), + incoming_info(GROUP, target, "AUTHOR-PLAY", 100), + )], + ) + .await; + let rows = store.messages(&jid(GROUP), None, 10).await.unwrap(); + assert_eq!(rows.len(), 2); + for row in rows { + assert_eq!( + row.status == MessageStatus::Played, + row.sender_jid == jid(target) + ); + assert_eq!( + store.audio_played_receipt_sent(&row).await.unwrap(), + row.sender_jid == jid(target) + ); + } +} + +#[tokio::test] +async fn pending_linked_plays_are_bounded_per_account() { + let (db, store) = test_store().await; + let db_b = db.database().store(2); + db.database().provision_device(2).await.unwrap(); + let other = ChatStore::new(&db_b).await.unwrap(); + feed( + &other, + [receipt( + jid(PEER), + jid(PEER), + &["OTHER-ACCOUNT"], + ReceiptType::PlayedSelf, + ts(100), + )], + ) + .await; + let ids: Vec = (0..4097).map(|id| format!("UNMATERIALIZED-{id}")).collect(); + feed( + &store, + ids.chunks(128).map(|chunk| { + let ids: Vec<_> = chunk.iter().map(String::as_str).collect(); + receipt(jid(PEER), jid(PEER), &ids, ReceiptType::PlayedSelf, ts(101)) + }), + ) + .await; + #[derive(diesel::QueryableByName)] + struct Count { + #[diesel(sql_type = diesel::sql_types::Integer)] + device_id: i32, + #[diesel(sql_type = diesel::sql_types::BigInt)] + count: i64, + } + let counts: Vec = db.shared().read(|conn| { + diesel::sql_query("SELECT device_id, count(*) AS count FROM audio_played_pending GROUP BY device_id ORDER BY device_id") + .load(conn).map_err(db_err) + }).await.unwrap(); + assert_eq!( + counts + .iter() + .map(|row| (row.device_id, row.count)) + .collect::>(), + vec![(1, 4096), (2, 1)] + ); +} + +#[tokio::test] +async fn authorless_linked_group_play_never_claims_the_first_materialized_author() { + let (_db, store) = test_store().await; + feed( + &store, + [receipt( + jid(GROUP), + jid(PEER), + &["SEPARATE-BATCHES"], + ReceiptType::PlayedSelf, + ts(101), + )], + ) + .await; + feed( + &store, + [message_event( + synthetic_voice(), + incoming_info(GROUP, PEER, "SEPARATE-BATCHES", 100), + )], + ) + .await; + let first = store + .message(&jid(GROUP), "SEPARATE-BATCHES") + .await + .unwrap() + .unwrap(); + assert_ne!(first.status, MessageStatus::Played); + assert!(!store.audio_played_receipt_sent(&first).await.unwrap()); + feed( + &store, + [message_event( + synthetic_voice(), + incoming_info( + GROUP, + "559900000002@s.whatsapp.net", + "SEPARATE-BATCHES", + 100, + ), + )], + ) + .await; + let rows = store.messages(&jid(GROUP), None, 10).await.unwrap(); + assert_eq!(rows.len(), 2); + for row in rows { + assert_ne!(row.status, MessageStatus::Played); + assert!(!store.audio_played_receipt_sent(&row).await.unwrap()); + } +} diff --git a/crates/core/src/message_status.rs b/crates/core/src/message_status.rs index 09c6f2a0..7d59fa6d 100644 --- a/crates/core/src/message_status.rs +++ b/crates/core/src/message_status.rs @@ -24,8 +24,11 @@ pub enum MessageStatus { Sent, /// It reached the recipient's device. Delivered, - /// The recipient opened the conversation, or played the voice note. + /// The recipient opened the conversation. Read, + /// Incoming audio was listened to here or on a linked device; outgoing + /// audio has a played receipt from its recipient. + Played, /// The send attempt failed. Terminal until the user retries, which starts /// a new message rather than moving this one. Failed, @@ -38,7 +41,10 @@ impl MessageStatus { /// failed send says something true about *this device*, and calling it /// read would claim a message was seen that never left. pub fn has_left_this_device(self) -> bool { - matches!(self, Self::Sent | Self::Delivered | Self::Read) + matches!( + self, + Self::Sent | Self::Delivered | Self::Read | Self::Played + ) } /// Move to `next` unless that would undo progress. @@ -68,7 +74,7 @@ impl MessageStatus { /// Whether the recipient has seen it — the state the ticks paint in the /// accent colour. pub fn is_read(self) -> bool { - matches!(self, Self::Read) + matches!(self, Self::Read | Self::Played) } /// A short label for assistive technology and tooltips, so the state is @@ -79,6 +85,7 @@ impl MessageStatus { Self::Sent => "Sent", Self::Delivered => "Delivered", Self::Read => "Read", + Self::Played => "Listened", Self::Failed => "Not sent", } } @@ -127,3 +134,19 @@ mod tests { assert_eq!(status, Delivered); } } + +#[cfg(test)] +mod played_tests { + use super::MessageStatus; + + #[test] + fn played_keeps_read_ticks_but_survives_stale_read_and_delivery() { + let mut status = MessageStatus::Read; + status.advance(MessageStatus::Played); + status.advance(MessageStatus::Read); + status.advance(MessageStatus::Delivered); + assert_eq!(status, MessageStatus::Played); + assert!(status.is_read()); + assert_eq!(status.label(), "Listened"); + } +} diff --git a/crates/daemon/src/server/requests.rs b/crates/daemon/src/server/requests.rs index d9608094..59628ee0 100644 --- a/crates/daemon/src/server/requests.rs +++ b/crates/daemon/src/server/requests.rs @@ -260,6 +260,9 @@ pub(super) async fn handle_request( ClientRequest::MarkRead(request) => { acted(dispatch(hub, commands, Action::MarkRead(request)).await) } + ClientRequest::MarkAudioPlayed(request) => { + acted(dispatch(hub, commands, Action::MarkAudioPlayed(request)).await) + } ClientRequest::MarkStatusWatched(request) => { acted(dispatch(hub, commands, Action::MarkStatusWatched(request)).await) } diff --git a/crates/daemon/src/server/tests.rs b/crates/daemon/src/server/tests.rs index 0c1fe0d2..0f877dbe 100644 --- a/crates/daemon/src/server/tests.rs +++ b/crates/daemon/src/server/tests.rs @@ -2472,6 +2472,46 @@ async fn replacing_an_account_runtime_closes_old_frontend_attachments() { line.clear(); assert_eq!(reader.read_line(&mut line).await.unwrap(), 0); } + +#[tokio::test] +async fn offline_and_terminal_audio_observations_reach_only_the_local_action() { + for state in [ + oxidezap_ipc::ConnectionState::Disconnected { + reason: "synthetic offline".into(), + }, + oxidezap_ipc::ConnectionState::LoggedOut { + message: "synthetic unlink".into(), + }, + ] { + let hub = StateHub::new(); + hub.apply(crate::state::Change::live( + oxidezap_ipc::DaemonEvent::ConnectionChanged(state), + )); + let (commands, taken) = bridge(CommandOutcome::Accepted); + let answer = handle_request( + bare(ClientRequest::MarkAudioPlayed( + oxidezap_ipc::MarkAudioPlayed { + jid: "559900000001@s.whatsapp.net".into(), + message_id: "SYNTHETIC-AUDIO".into(), + }, + )), + &hub, + &no_plugins(), + &commands, + &outbox(), + ) + .await; + assert!(matches!( + parse(answer.frame), + DaemonMessage::Accepted { .. } + )); + let action = taken.await.unwrap().unwrap(); + assert!(matches!(action, Action::MarkAudioPlayed(_))); + assert!(!action.needs_network()); + assert!(!hub.connection().is_connected()); + } +} + #[tokio::test] async fn scoped_wire_hello_confirms_successor_and_never_rebinds_after_selection_changes() { use oxidezap_wire::{ diff --git a/crates/daemon/src/session_bridge/act.rs b/crates/daemon/src/session_bridge/act.rs index 37108038..b14a961c 100644 --- a/crates/daemon/src/session_bridge/act.rs +++ b/crates/daemon/src/session_bridge/act.rs @@ -103,6 +103,29 @@ impl Bridge { reply: tokio::sync::oneshot::Sender, ) -> Option<(Action, tokio::sync::oneshot::Sender)> { match action { + Action::MarkAudioPlayed(request) => { + let Some(permit) = self.permit() else { + let _ = reply.send(too_busy()); + return None; + }; + let hub = self.hub.clone(); + let lifecycle = self.lifecycle.clone(); + let task = client.mark_audio_played(request.jid, request.message_id, move || { + hub.connection().is_connected() + && !hub.is_logged_out() + && !lifecycle.is_stopping() + }); + oxidezap_session::spawn(async move { + let outcome = match task.await { + Ok(Ok(())) => CommandOutcome::Accepted, + Ok(Err(error)) => CommandOutcome::Refused(error), + Err(error) => CommandOutcome::Refused(error.to_string()), + }; + let _ = reply.send(outcome); + drop(permit); + }); + None + } Action::EditMessage { id, request, @@ -1611,7 +1634,8 @@ impl Bridge { through_message_id, }) => self.mark_read(client, &jid, through_message_id.as_deref()), // Answered from a task of its own. See `begin_slow`. - Action::MarkStatusWatched(_) + Action::MarkAudioPlayed(_) + | Action::MarkStatusWatched(_) | Action::LoadMessages { .. } | Action::LoadChats { .. } | Action::GroupMembers { .. } => { diff --git a/crates/daemon/src/session_bridge/action.rs b/crates/daemon/src/session_bridge/action.rs index ddb02baf..4dcdc195 100644 --- a/crates/daemon/src/session_bridge/action.rs +++ b/crates/daemon/src/session_bridge/action.rs @@ -38,6 +38,7 @@ pub enum Action { SendMedia(oxidezap_ipc::SendMedia), SendReaction(oxidezap_ipc::SendReaction), MarkRead(oxidezap_ipc::MarkRead), + MarkAudioPlayed(oxidezap_ipc::MarkAudioPlayed), MarkStatusWatched(oxidezap_ipc::MarkStatusWatched), Typing(oxidezap_ipc::Typing), Call(CallAction), @@ -234,6 +235,7 @@ impl Action { | Self::ReconnectSession | Self::RefreshVideo | Self::ForgetSession(_) + | Self::MarkAudioPlayed(_) | Self::MarkStatusWatched(_) | Self::LoadMessages { .. } | Self::LoadChats { .. } diff --git a/crates/daemon/src/session_bridge/wire_events.rs b/crates/daemon/src/session_bridge/wire_events.rs index 053a4af1..78edbc45 100644 --- a/crates/daemon/src/session_bridge/wire_events.rs +++ b/crates/daemon/src/session_bridge/wire_events.rs @@ -171,6 +171,7 @@ pub(crate) fn chat_message_to_dto(chat_jid: &str, msg: ChatMessage) -> MessageDt oxidezap_core::MessageStatus::Sent => "sent", oxidezap_core::MessageStatus::Delivered => "delivered", oxidezap_core::MessageStatus::Read => "read", + oxidezap_core::MessageStatus::Played => "played", oxidezap_core::MessageStatus::Failed => "failed", }; diff --git a/crates/gui/src/app/commands.rs b/crates/gui/src/app/commands.rs index 2f84c629..712c5010 100644 --- a/crates/gui/src/app/commands.rs +++ b/crates/gui/src/app/commands.rs @@ -101,7 +101,7 @@ impl WhatsAppApp { self.viewer.update(cx, |state, cx| state.open(viewer, cx)); // A photo and a voice note both want the speakers; opening one stops // the other rather than talking over it. - self.stop_current_media(); + self.stop_user_media(); // The arrow keys follow in `sync_overlay_focus`, on this same frame — // and, unlike focusing here, they are handed back when the viewer // goes away, however it goes away. diff --git a/crates/gui/src/app/media_ctl.rs b/crates/gui/src/app/media_ctl.rs index 5c285023..7a505842 100644 --- a/crates/gui/src/app/media_ctl.rs +++ b/crates/gui/src/app/media_ctl.rs @@ -44,6 +44,17 @@ impl PendingAudioPreparation { } } +fn audio_receipt_chat(chats: &[Arc], id: &str) -> Option { + let mut rows = chats.iter().flat_map(|chat| { + chat.messages + .iter() + .filter(move |message| message.id == id) + .map(move |message| (chat, message)) + }); + let (chat, message) = rows.next()?; + (!message.is_from_me && rows.next().is_none()).then(|| chat.jid.clone()) +} + fn media_identity(data: &Arc>) -> usize { Arc::as_ptr(data) as usize } @@ -225,7 +236,13 @@ impl WhatsAppApp { self.sweep_retained_media_keeping(Some(message_id), cx); } } - /// Stop any currently playing media. Does NOT call cx.notify(). + /// Preserve a successful short play before a user switch clears its evidence. + pub(super) fn stop_user_media(&mut self) { + self.observe_audio_playback(); + self.stop_current_media(); + } + + /// Discard playback during teardown or failed preparation, without notifying the UI. pub(super) fn stop_current_media(&mut self) { self.audio_player.stop(); // A stopped tick may still be between its sleep and `entity.update`. @@ -266,7 +283,7 @@ impl WhatsAppApp { /// must not start sound behind the chat list, Settings, or Status. pub(super) fn cancel_hidden_media_autoplay(&mut self) { if self.audio_preparation.is_some() { - self.stop_current_media(); + self.stop_user_media(); } else { self.pending_media_request = None; } @@ -436,18 +453,36 @@ impl WhatsAppApp { /// pass and left the button, the clock and the waveform frozen for the /// whole note. `is_loading` is that gap, and it is false everywhere the /// decoder is synchronous. + fn observe_audio_playback(&mut self) { + if !self.audio_player.take_started() { + return; + } + let ActiveMedia::Audio { message_id } = &self.active_media else { + return; + }; + let Some(chat) = audio_receipt_chat(&self.chats, message_id) else { + return; + }; + if let Some(client) = &self.client { + client.mark_audio_played(chat, message_id.clone()); + } + } + pub(super) fn ensure_playback_tick(&mut self, cx: &mut Context) { if self.playback_tick.is_some() { return; } let epoch = self.playback_epoch; + let account = self.client.as_ref().map(|client| client.account()); self.playback_tick = Some(cx.spawn(async move |entity: WeakEntity, cx| { loop { // ~15fps: the playhead only has to look continuous, and a // voice note is not worth a frame-rate repaint of the list. crate::platform::sleep(std::time::Duration::from_millis(66)).await; let running = entity.update(cx, |app, cx| { - if app.playback_epoch != epoch { + if app.playback_epoch != epoch + || app.client.as_ref().map(|client| client.account()) != account + { return false; } // The browser's `play_prepared` returns before @@ -458,6 +493,7 @@ impl WhatsAppApp { if !app.audio_player.is_loading() { app.audio_preparation = None; } + app.observe_audio_playback(); let playing = app.audio_player.is_playing(); if playing { cx.notify(); @@ -499,7 +535,7 @@ impl WhatsAppApp { was_playing: bool, cx: &mut Context, ) { - self.stop_current_media(); + self.stop_user_media(); self.pending_media_request = Some(message_id.clone()); self.audio_player.set_speed(self.playback_speed); // On the web this must happen synchronously while the click still has @@ -572,6 +608,7 @@ impl WhatsAppApp { if app.playback_epoch == completion_epoch && app.active_media.is_playing(&completed_id) { + app.observe_audio_playback(); app.active_media = ActiveMedia::None; info!("Audio playback completed"); cx.notify(); @@ -723,7 +760,7 @@ impl WhatsAppApp { // Stop any current playback, video included: a playing video must not // keep running underneath the download. - self.stop_current_media(); + self.stop_user_media(); self.pending_media_request = Some(message_id.clone()); let msg_id = message_id.clone(); @@ -1296,7 +1333,7 @@ impl WhatsAppApp { // this video's own paused audio and it would come back mute. let owns_audio = self.audio.message_id() == Some(message_id.as_str()); if !self.active_media.is_playing(&message_id) && !owns_audio { - self.stop_current_media(); + self.stop_user_media(); } self.pending_media_request = Some(message_id.clone()); @@ -1407,7 +1444,7 @@ impl WhatsAppApp { let playing = self.playing_video_id().map(str::to_string); // Stop any currently playing media (mutual exclusion) - self.stop_current_media(); + self.stop_user_media(); self.pending_media_request = Some(message_id.clone()); // Evict old video players if the cache is full, sparing the one that @@ -1650,6 +1687,32 @@ async fn hand_to_user( mod tests { use super::*; + #[test] + fn audio_receipt_identity_refuses_own_or_cross_chat_id_collisions() { + let mut first = Chat::new("559900000001@s.whatsapp.net".into()); + first.messages.push(ChatMessage::new_incoming( + "VOICE".into(), + "sender".into(), + String::new(), + )); + assert_eq!( + audio_receipt_chat(&[Arc::new(first.clone())], "VOICE"), + Some(first.jid.clone()) + ); + let mut second = Chat::new("111000011112222@lid".into()); + second + .messages + .push(ChatMessage::new_outgoing("VOICE".into(), String::new())); + assert_eq!( + audio_receipt_chat(&[Arc::new(second.clone())], "VOICE"), + None + ); + assert_eq!( + audio_receipt_chat(&[Arc::new(first), Arc::new(second)], "VOICE"), + None + ); + } + fn pending(epoch: usize, position: f32, was_playing: bool) -> PendingAudioPreparation { PendingAudioPreparation { message_id: "voice-1".to_string(), diff --git a/crates/gui/src/app/mod.rs b/crates/gui/src/app/mod.rs index 5a28ec26..f23200ac 100644 --- a/crates/gui/src/app/mod.rs +++ b/crates/gui/src/app/mod.rs @@ -2851,7 +2851,7 @@ impl WhatsAppApp { // not to the next list reusing the window selection layer. clear_window_message_selection(window, cx); } - self.stop_current_media(); + self.stop_user_media(); // Leaving a chat mid-composition: release its typing indicator now, // or it would stay "typing..." and the eventual paused would land on // the newly selected chat instead. @@ -3829,10 +3829,14 @@ impl WhatsAppApp { let Some(chat) = self.find_chat_mut(&chat_jid) else { return; }; - let advanced = chat.advance_status(&message_ids, status); + let advanced = if receipt_type == ReceiptType::PlayedSelf { + advance_authorless_self_play(chat, &chat_jid, &message_ids) + } else { + chat.advance_status(&message_ids, status) + }; // A read receipt is also the peer telling us they opened the chat, // which clears our own unread state for the messages named. - let read = if status == MessageStatus::Read { + let read = if status.is_read() { chat.mark_messages_as_read(&message_ids) } else { 0 @@ -4047,12 +4051,38 @@ fn receipt_status(receipt_type: ReceiptType) -> Option { ReceiptType::Delivered => Some(MessageStatus::Delivered), // Played is Read plus "and listened to it"; the ticks are the same. ReceiptType::Read | ReceiptType::ReadSelf => Some(MessageStatus::Read), - ReceiptType::Played | ReceiptType::PlayedSelf => Some(MessageStatus::Read), + ReceiptType::Played | ReceiptType::PlayedSelf => Some(MessageStatus::Played), // Retries, errors and sender echoes say nothing about delivery. _ => None, } } +// The live event has no original-author field. Participant-keyed plays must +// arrive through confirmed store hydration instead of a currently unique row. +fn advance_authorless_self_play(chat: &mut Chat, chat_jid: &str, ids: &[String]) -> usize { + let Ok(jid) = chat_jid.parse::() else { + return 0; + }; + if !(jid.is_pn() || jid.is_lid() || jid.is_newsletter()) { + return 0; + } + let mut changed = 0; + for id in ids { + let mut matches = chat + .messages + .iter_mut() + .filter(|message| !message.is_from_me && message.id == *id); + if let Some(message) = matches.next() + && matches.next().is_none() + { + let before = message.status; + message.status.advance(MessageStatus::Played); + changed += usize::from(before != message.status); + } + } + changed +} + /// The newest message in `chat` that the daemon can also name. /// /// `MarkRead` is a claim about what the requester saw, and the daemon checks it @@ -5957,3 +5987,42 @@ mod decoded_image_budget { assert!(over_budget(far_over, MIN_DECODED_IMAGES + 1)); } } + +#[cfg(test)] +mod linked_play_tests { + use super::*; + + #[test] + fn live_authorless_group_play_never_claims_a_loaded_row() { + let mut chat = Chat::new("120363000000000001@g.us".into()); + let mut incoming = ChatMessage::new_incoming( + "same-id".into(), + "5511000000002@s.whatsapp.net".into(), + "voice".into(), + ); + incoming.status = MessageStatus::Read; + chat.messages.push(incoming); + assert_eq!( + advance_authorless_self_play(&mut chat, "120363000000000001@g.us", &["same-id".into()]), + 0 + ); + assert_eq!(chat.messages[0].status, MessageStatus::Read); + assert_eq!( + advance_authorless_self_play( + &mut chat, + "5511000000002@s.whatsapp.net", + &["same-id".into()] + ), + 1 + ); + assert_eq!(chat.messages[0].status, MessageStatus::Played); + assert_eq!( + advance_authorless_self_play( + &mut chat, + "5511000000002@s.whatsapp.net", + &["same-id".into()] + ), + 0 + ); + } +} diff --git a/crates/gui/src/components/message_bubble/audio.rs b/crates/gui/src/components/message_bubble/audio.rs index 0d90e86c..a3eabd1d 100644 --- a/crates/gui/src/components/message_bubble/audio.rs +++ b/crates/gui/src/components/message_bubble/audio.rs @@ -79,6 +79,7 @@ impl AudioLoadState { pub(super) fn render_audio_player( media_content: MediaContent, message_id: String, + listened: Option<&'static str>, is_playing: bool, // Progress only means anything for the clip that is actually loaded; a // second voice note in the same conversation must not borrow its @@ -142,6 +143,9 @@ pub(super) fn render_audio_player( // Counts up while playing, shows the total at rest — the same // number a listener wants at each moment. .child(format_clock(elapsed.or(duration))) + .when_some(listened, |el, label| { + el.child(div().text_color(cx.theme().ring).child(label)) + }) .children(state.status().map(|status| { div() .text_size(metrics.text_micro()) @@ -516,3 +520,32 @@ mod tests { assert_eq!(format_clock(Some(14)), "0:14"); } } + +/// A read tick never means audio was heard; direction names whose play it was. +pub(super) fn played_label( + from_me: bool, + status: oxidezap_core::MessageStatus, +) -> Option<&'static str> { + (status == oxidezap_core::MessageStatus::Played).then_some(if from_me { + "Played by recipient" + } else { + "Listened" + }) +} + +#[cfg(test)] +mod played_indicator_tests { + use super::*; + use oxidezap_core::MessageStatus; + + #[test] + fn indicator_distinguishes_local_or_linked_play_from_recipient_receipt() { + assert_eq!(played_label(false, MessageStatus::Read), None); + assert_eq!(played_label(true, MessageStatus::Read), None); + assert_eq!(played_label(false, MessageStatus::Played), Some("Listened")); + assert_eq!( + played_label(true, MessageStatus::Played), + Some("Played by recipient") + ); + } +} diff --git a/crates/gui/src/components/message_bubble/media.rs b/crates/gui/src/components/message_bubble/media.rs index d4148bf8..c0331f51 100644 --- a/crates/gui/src/components/message_bubble/media.rs +++ b/crates/gui/src/components/message_bubble/media.rs @@ -33,6 +33,7 @@ pub(super) struct MediaProps { pub decoded_image: Option>, pub audio: Option, pub playback_speed: f32, + pub listened: Option<&'static str>, pub is_downloading: bool, pub is_preparing: bool, pub saved_document_path: Option, @@ -54,6 +55,7 @@ pub(super) fn render_media_content( decoded_image, audio, playback_speed, + listened, is_downloading, is_preparing, saved_document_path, @@ -260,6 +262,7 @@ pub(super) fn render_media_content( MediaType::Audio => el.child(super::audio::render_audio_player( media_content, message_id, + listened, is_playing, audio, playback_speed, diff --git a/crates/gui/src/components/message_bubble/mod.rs b/crates/gui/src/components/message_bubble/mod.rs index 5bd1a024..2cb3bfa6 100644 --- a/crates/gui/src/components/message_bubble/mod.rs +++ b/crates/gui/src/components/message_bubble/mod.rs @@ -302,6 +302,10 @@ pub fn render_message_bubble( decoded_image: props.decoded_image.clone(), audio: props.audio, playback_speed: props.playback_speed, + listened: audio::played_label( + message.is_from_me, + message.status, + ), is_downloading: props.is_downloading, is_preparing: props.is_preparing, saved_document_path: props.saved_document_path.clone(), diff --git a/crates/gui/src/components/status_ticks.rs b/crates/gui/src/components/status_ticks.rs index bcb7f09a..75e57113 100644 --- a/crates/gui/src/components/status_ticks.rs +++ b/crates/gui/src/components/status_ticks.rs @@ -60,7 +60,9 @@ fn ticks( MessageStatus::Delivered => (ProductIcon::CheckCheck.into(), quiet), // The one state that earns colour: it is the answer to "did they see // it", which is the question the ticks exist for. - MessageStatus::Read => (ProductIcon::CheckCheck.into(), cx.theme().ring), + MessageStatus::Read | MessageStatus::Played => { + (ProductIcon::CheckCheck.into(), cx.theme().ring) + } MessageStatus::Failed => (Icon::new(IconName::TriangleAlert), cx.theme().danger), }; diff --git a/crates/gui/src/session/mod.rs b/crates/gui/src/session/mod.rs index 4a6f9cf7..607fd780 100644 --- a/crates/gui/src/session/mod.rs +++ b/crates/gui/src/session/mod.rs @@ -1345,6 +1345,13 @@ impl SessionHandle { }))); } + /// Report playback after the audio sink advances. + pub fn mark_audio_played(&self, jid: String, message_id: String) { + self.tell(ClientRequest::MarkAudioPlayed( + oxidezap_ipc::MarkAudioPlayed { jid, message_id }, + )); + } + /// Mark a chat read up to the message the UI is looking at. /// /// One request where the old client took two: the daemon owns the read diff --git a/crates/ipc/src/lib.rs b/crates/ipc/src/lib.rs index cd1f88e4..60e77bb5 100644 --- a/crates/ipc/src/lib.rs +++ b/crates/ipc/src/lib.rs @@ -55,9 +55,10 @@ pub use protocol::{ AccountIdentity, AccountOverview, AccountStatus, AccountsSnapshot, AvatarDemand, CallAction, CallDeviceAction, CallDeviceReply, CameraPreviewFrame, ChatSummary, ClientRequest, ClientScope, ConnectionState, DaemonEvent, DaemonMessage, Download, EditMessage, EnsureAvatars, - GroupMembers, InstallPlugin, LoadChats, LoadMessages, MarkRead, MarkStatusWatched, - MessagePreview, PageCursor, PairingCode, ProtocolError, Request, RequestId, RevokeMessage, - SendAudio, SendMedia, SendReaction, SendText, StateSnapshot, StateVersion, Typing, VotePoll, + GroupMembers, InstallPlugin, LoadChats, LoadMessages, MarkAudioPlayed, MarkRead, + MarkStatusWatched, MessagePreview, PageCursor, PairingCode, ProtocolError, Request, RequestId, + RevokeMessage, SendAudio, SendMedia, SendReaction, SendText, StateSnapshot, StateVersion, + Typing, VotePoll, }; pub use transport::{ ACCOUNT_STAGED_INFIX, DEFAULT_WEB_PORT, MAX_STAGED_BYTES, PROTOCOL_VERSION, STAGED_PREFIX, diff --git a/crates/ipc/src/protocol.rs b/crates/ipc/src/protocol.rs index 14dcb4da..51b786b9 100644 --- a/crates/ipc/src/protocol.rs +++ b/crates/ipc/src/protocol.rs @@ -856,6 +856,14 @@ pub struct MarkRead { pub through_message_id: Option, } +/// A successful audio playback observed by a front end. Routing and media +/// eligibility are checked against the account store. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct MarkAudioPlayed { + pub jid: String, + pub message_id: String, +} + /// Status updates that have been watched. See /// [`ClientRequest::MarkStatusWatched`]. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -1047,6 +1055,8 @@ pub enum ClientRequest { /// when this is not the message it would name in a snapshot right now, and /// the client resends after catching up. MarkRead(MarkRead), + /// Record actual playback; offline observations are local only. + MarkAudioPlayed(MarkAudioPlayed), /// Remember that these status updates have been watched. /// /// Not a [`MarkRead`](Self::MarkRead) for the broadcast: that one clears a @@ -2377,3 +2387,23 @@ mod device_contract_tests { ); } } + +#[cfg(test)] +mod audio_played_tests { + use super::*; + + #[test] + fn playback_observation_carries_identity_without_frontend_routing_claims() { + let request = ClientRequest::MarkAudioPlayed(MarkAudioPlayed { + jid: "120363000000000001@g.us".into(), + message_id: "SYNTHETIC-AUDIO".into(), + }); + let json = serde_json::to_string(&request).unwrap(); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + request + ); + assert!(json.contains("mark_audio_played")); + assert!(!json.contains("participant")); + } +} diff --git a/crates/session/Cargo.toml b/crates/session/Cargo.toml index 1cb7ad72..faa54d75 100644 --- a/crates/session/Cargo.toml +++ b/crates/session/Cargo.toml @@ -162,6 +162,7 @@ wasm-bindgen-test = "0.3" # runs on the host, where `cargo test` already is. Only the test needs these # here; the page's build gets them from the wasm section above. [dev-dependencies] +anyhow = { workspace = true } diesel = { workspace = true } futures-channel = "0.3" futures-lite = "2" diff --git a/crates/session/src/whatsapp/convert.rs b/crates/session/src/whatsapp/convert.rs index b9939607..93c5d632 100644 --- a/crates/session/src/whatsapp/convert.rs +++ b/crates/session/src/whatsapp/convert.rs @@ -101,7 +101,7 @@ pub(super) fn stored_to_chat_message(stored: oxidezap_chat_store::StoredMessage) // The store has tracked the real delivery state all along; the UI used // to flatten it to a bool and lose the delivered/read distinction that // the second tick exists to show. - status: if stored.from_me { + status: if stored.from_me || stored.status == oxidezap_chat_store::MessageStatus::Played { store_status(stored.status) } else { MessageStatus::default() @@ -165,8 +165,8 @@ fn store_status(status: oxidezap_chat_store::MessageStatus) -> MessageStatus { Stored::Pending => MessageStatus::Pending, Stored::ServerAck => MessageStatus::Sent, Stored::Delivered => MessageStatus::Delivered, - // Played is Read plus "and listened to it"; the ticks are the same. - Stored::Read | Stored::Played => MessageStatus::Read, + Stored::Read => MessageStatus::Read, + Stored::Played => MessageStatus::Played, } } @@ -487,6 +487,22 @@ mod tests { "quoted-message" ); } + #[test] + fn incoming_played_survives_hydration_without_turning_read_into_played() { + let mut stored = stored_poll_creation(); + stored.kind = oxidezap_chat_store::MessageKind::Audio; + for from_me in [false, true] { + stored.from_me = from_me; + stored.status = oxidezap_chat_store::MessageStatus::Played; + assert_eq!( + stored_to_chat_message(stored.clone()).status, + MessageStatus::Played + ); + } + stored.from_me = false; + stored.status = oxidezap_chat_store::MessageStatus::Read; + assert_ne!(stored_to_chat_message(stored).status, MessageStatus::Played); + } } #[cfg(test)] diff --git a/crates/session/src/whatsapp/mod.rs b/crates/session/src/whatsapp/mod.rs index 0e2d6851..3db69c70 100644 --- a/crates/session/src/whatsapp/mod.rs +++ b/crates/session/src/whatsapp/mod.rs @@ -36,6 +36,7 @@ mod shared_history; /// What a message's media is, for both the roads it arrives on. mod media; +mod played; /// What a picked file becomes on the wire: its shape, and the message that /// carries it. @@ -687,6 +688,7 @@ pub struct WhatsAppClient { /// session is alive, so an async completion cannot accidentally write to a /// different account after a front-end switch. account_id: AccountId, + played_gate: Arc>>, /// Shared registry for the physical database and one chat migration run. stores: Arc, /// Where the session's work runs: a runtime on a thread of its own on a @@ -797,6 +799,7 @@ impl WhatsAppClient { let (ui_sender, ui_events) = ui_queue::channel(reload.clone(), history_budget.clone()); Ok(Self { account_id, + played_gate: Arc::new(std::sync::Mutex::new(HashSet::new())), stores, exec: Executor::new()?, ui_sender, @@ -2019,12 +2022,7 @@ impl WhatsAppClient { // Delivered used to be dropped here, which is why the // second tick never appeared: only Read and Played reached // the UI, so a message went from one tick straight to blue. - let Some(dominated_type) = (match &receipt.r#type { - ReceiptType::Delivered => Some(ReceiptType::Delivered), - ReceiptType::Read | ReceiptType::ReadSelf => Some(ReceiptType::Read), - ReceiptType::Played | ReceiptType::PlayedSelf => Some(ReceiptType::Played), - _ => None, - }) else { + let Some(dominated_type) = played::receipt_for_ui(&receipt.r#type) else { return; }; diff --git a/crates/session/src/whatsapp/played.rs b/crates/session/src/whatsapp/played.rs new file mode 100644 index 00000000..6f2fd0c7 --- /dev/null +++ b/crates/session/src/whatsapp/played.rs @@ -0,0 +1,505 @@ +//! Playback observations belong to the captured account session, never a successor. + +use super::*; +use oxidezap_chat_store::{MessageKind, StoredMessage}; +use whatsapp_rust::{MessageId, MessageRef}; + +pub(super) fn receipt_for_ui(receipt: &ReceiptType) -> Option { + match receipt { + ReceiptType::Delivered => Some(ReceiptType::Delivered), + ReceiptType::Read | ReceiptType::ReadSelf => Some(ReceiptType::Read), + ReceiptType::Played => Some(ReceiptType::Played), + ReceiptType::PlayedSelf => Some(ReceiptType::PlayedSelf), + _ => None, + } +} + +fn eligible(message: &StoredMessage) -> bool { + if message.from_me || message.revoked { + return false; + } + match message.kind { + MessageKind::VoiceNote => true, + MessageKind::Audio => { + message.chat_jid.is_newsletter() + || message + .message + .as_deref() + .is_some_and(MessageExt::is_view_once) + || message + .message + .as_deref() + .map(oxidezap_chat_store::normalized_message) + .and_then(|m| m.audio_message.as_option()) + .and_then(|m| m.context_info.as_option()) + .is_some_and(|c| c.forwarded_newsletter_message_info.is_set()) + } + _ => false, + } +} + +impl WhatsAppClient { + /// Offline plays are durable local observations, with no deferred network job. + /// A later real play can retry a failed transport send. Success here means + /// transport submission, not server acknowledgement or another device updating. + pub fn mark_audio_played( + &self, + chat: String, + id: String, + send_allowed: impl Fn() -> bool + Send + Sync + 'static, + ) -> Task> { + // Capture now: a task first polled after teardown must not select a new session. + let live = self.session.try_lock().ok().and_then(|slot| slot.clone()); + let gate = self.played_gate.clone(); + let send = send_allowed(); + self.exec.spawn(async move { + let live = live.ok_or("No account session")?; + let chat: Jid = chat.parse().map_err(|e| format!("Invalid chat: {e}"))?; + let message = live + .chat_store + .message(&chat, &id) + .await + .map_err(|e| e.to_string())? + .ok_or("Audio message is not stored")?; + if !eligible(&message) { + return Ok(()); + } + let _guard = InFlight::claim(gate, message.seq)?; + observe_playback( + &live.chat_store, + &message, + send.then_some(|| async { + let chat: Jid = live + .names + .chat_key(&live.client, &message.chat_jid) + .await + .ok_or("Could not resolve the receipt's PN/LID address")? + .parse() + .map_err(|e| format!("Invalid receipt chat: {e}"))?; + let participant = participant_keyed_chat(&chat).then_some(&message.sender_jid); + if !send_allowed() { + return Err("WhatsApp went offline; playback was recorded locally".into()); + } + send_played(&live.client, &chat, participant, &message.id).await + }), + ) + .await + }) + } +} + +async fn send_played( + client: &Client, + chat: &Jid, + participant: Option<&Jid>, + id: &str, +) -> Result<(), String> { + // MessageRef accepts user/group chats. The existing raw API also supports newsletters. + if chat.is_newsletter() { + return client + .mark_as_played(chat, None, &[id]) + .await + .map_err(|e| e.to_string()); + } + let target = MessageRef::new( + chat, + MessageId::new(id).map_err(|e| e.to_string())?, + participant, + false, + ) + .map_err(|e| e.to_string())?; + client + .mark_message_played(&target) + .await + .map_err(|e| e.to_string()) +} + +struct InFlight { + keys: Arc>>, + row: i64, +} +impl InFlight { + fn claim(keys: Arc>>, row: i64) -> Result { + let inserted = keys + .lock() + .map_err(|_| "Playback tracker unavailable")? + .insert(row); + if !inserted { + return Err("Playback observation already in flight".into()); + } + Ok(Self { keys, row }) + } +} +impl Drop for InFlight { + fn drop(&mut self) { + if let Ok(mut keys) = self.keys.lock() { + keys.remove(&self.row); + } + } +} + +async fn observe_playback( + store: &ChatStore, + message: &StoredMessage, + send: Option, +) -> Result<(), String> +where + F: FnOnce() -> Fut, + Fut: std::future::Future>, +{ + if !eligible(message) { + return Ok(()); + } + store + .record_audio_played(message, false) + .map_err(|e| e.to_string())?; + store.flush().await.map_err(|e| e.to_string())?; + let Some(send) = send else { + return Ok(()); + }; + if store + .audio_played_receipt_sent(message) + .await + .map_err(|e| e.to_string())? + { + return Ok(()); + } + send().await?; + store + .record_audio_played(message, true) + .map_err(|e| e.to_string())?; + store.flush().await.map_err(|e| e.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use whatsapp_rust::wacore::types::events::InboundMessage; + use whatsapp_rust::wacore::types::message::{MessageInfo, MessageSource}; + + #[test] + fn linked_play_is_not_reinterpreted_as_a_recipient_play() { + assert_eq!( + receipt_for_ui(&ReceiptType::PlayedSelf), + Some(ReceiptType::PlayedSelf) + ); + assert_eq!( + receipt_for_ui(&ReceiptType::Played), + Some(ReceiptType::Played) + ); + } + + async fn fixture(group: bool, ptt: bool) -> (SqliteStore, Arc, StoredMessage) { + static NEXT: AtomicUsize = AtomicUsize::new(0); + let db = SqliteStore::open(&format!( + "file:audio_played_{}_{}?mode=memory&cache=shared", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )) + .await + .unwrap(); + db.database() + .provision_device(db.device_id()) + .await + .unwrap(); + let store = ChatStore::new(&db).await.unwrap(); + let chat: Jid = if group { + "120363000000000001@g.us" + } else { + "559900000001@s.whatsapp.net" + } + .parse() + .unwrap(); + let sender: Jid = "111000011112222@lid".parse().unwrap(); + let mut audio = wa::message::AudioMessage::default(); + audio.ptt = Some(ptt); + let mut proto = wa::Message::default(); + proto.audio_message = whatsapp_rust::buffa::MessageField::some(audio); + let info = MessageInfo { + source: MessageSource { + chat: chat.clone(), + sender, + is_group: group, + ..Default::default() + }, + id: "AUDIO-SYNTHETIC".into(), + ..Default::default() + }; + store + .commit_inbound_batch(&[InboundMessage::builder() + .message(Arc::new(proto)) + .info(Arc::new(info)) + .build()]) + .await + .unwrap(); + let message = store + .message(&chat, "AUDIO-SYNTHETIC") + .await + .unwrap() + .unwrap(); + (db, store, message) + } + + #[tokio::test] + async fn actual_observation_retries_failure_but_never_repeats_success() { + let (_db, store, message) = fixture(false, true).await; + let mut calls = 0; + let failed = observe_playback( + &store, + &message, + Some(|| { + calls += 1; + std::future::ready(Err("transport failed".into())) + }), + ) + .await; + assert!(failed.is_err()); + assert!(!store.audio_played_receipt_sent(&message).await.unwrap()); + for _ in 0..3 { + observe_playback( + &store, + &message, + Some(|| { + calls += 1; + std::future::ready(Ok(())) + }), + ) + .await + .unwrap(); + } + assert_eq!(calls, 2); + assert_eq!( + store + .message(&message.chat_jid, &message.id) + .await + .unwrap() + .unwrap() + .status, + oxidezap_chat_store::MessageStatus::Played + ); + } + + #[tokio::test] + async fn offline_play_is_local_and_other_account_never_inherits_submission() { + let (_db, store, message) = fixture(false, true).await; + observe_playback( + &store, + &message, + None:: std::future::Ready>>, + ) + .await + .unwrap(); + assert!(!store.audio_played_receipt_sent(&message).await.unwrap()); + assert_eq!( + store + .message(&message.chat_jid, &message.id) + .await + .unwrap() + .unwrap() + .status, + oxidezap_chat_store::MessageStatus::Played + ); + let (_other_db, other_store, other_message) = fixture(false, true).await; + assert_ne!( + other_store + .message(&other_message.chat_jid, &other_message.id) + .await + .unwrap() + .unwrap() + .status, + oxidezap_chat_store::MessageStatus::Played + ); + } + + #[tokio::test] + async fn played_incoming_survives_store_reopen_and_ui_hydration() { + let (db, store, message) = fixture(false, true).await; + observe_playback( + &store, + &message, + None:: std::future::Ready>>, + ) + .await + .unwrap(); + drop(store); + let reopened = ChatStore::new(&db).await.unwrap(); + let row = reopened + .message(&message.chat_jid, &message.id) + .await + .unwrap() + .unwrap(); + let hydrated = super::super::convert::stored_to_chat_message(row); + assert!(!hydrated.is_from_me); + assert_eq!(hydrated.status, oxidezap_core::MessageStatus::Played); + } + + #[tokio::test] + async fn own_and_ordinary_audio_do_not_emit_receipts() { + let (_db, store, mut message) = fixture(false, true).await; + message.from_me = true; + observe_playback( + &store, + &message, + Some(|| async { panic!("own audio receipt") }), + ) + .await + .unwrap(); + let (_db2, store, message) = fixture(false, false).await; + observe_playback( + &store, + &message, + Some(|| async { panic!("ordinary audio receipt") }), + ) + .await + .unwrap(); + assert_ne!( + store + .message(&message.chat_jid, &message.id) + .await + .unwrap() + .unwrap() + .status, + oxidezap_chat_store::MessageStatus::Played + ); + } + + #[tokio::test] + async fn newsletter_origin_and_view_once_audio_follow_the_official_eligibility_guards() { + let (_db, _store, mut message) = fixture(false, false).await; + assert!(!eligible(&message)); + let mut proto = message.message.take().unwrap(); + proto.audio_message.as_option_mut().unwrap().view_once = Some(true); + message.message = Some(proto); + assert!(eligible(&message)); + let mut proto = message.message.take().unwrap(); + let audio = proto.audio_message.as_option_mut().unwrap(); + audio.view_once = None; + let mut context = wa::ContextInfo::default(); + context.forwarded_newsletter_message_info = + whatsapp_rust::buffa::MessageField::some(Default::default()); + audio.context_info = whatsapp_rust::buffa::MessageField::some(context); + message.message = Some(proto); + assert!(eligible(&message)); + message.from_me = true; + assert!(!eligible(&message)); + message.from_me = false; + message.message = None; + message.chat_jid = "120363000000000002@newsletter".parse().unwrap(); + assert!(eligible(&message)); + message.revoked = true; + assert!(!eligible(&message)); + } + + #[tokio::test] + async fn group_uses_original_participant_and_direct_drops_it() { + for group in [false, true] { + let (_db, _store, message) = fixture(group, true).await; + let id = MessageId::new(message.id.clone()).unwrap(); + let participant = + participant_keyed_chat(&message.chat_jid).then_some(&message.sender_jid); + let target = MessageRef::new(&message.chat_jid, id, participant, false).unwrap(); + assert_eq!( + target.receipt_sender(), + group.then_some(&message.sender_jid) + ); + assert_eq!(target.chat(), &message.chat_jid); + } + } + + #[test] + fn multiple_windows_coalesce_and_failure_releases_the_key() { + let keys = Arc::new(std::sync::Mutex::new(HashSet::new())); + let first = InFlight::claim(keys.clone(), 1).unwrap(); + assert!(InFlight::claim(keys.clone(), 1).is_err()); + assert!(InFlight::claim(keys.clone(), 2).is_ok()); + drop(first); + assert!(InFlight::claim(keys, 1).is_ok()); + } +} + +#[cfg(all(test, not(target_family = "wasm")))] +mod wire_tests { + use super::*; + use whatsapp_rust::client::NodeFilter; + use whatsapp_rust::store::persistence_manager::PersistenceManager; + use whatsapp_rust::wacore::net::{ + HttpClient, HttpRequest, HttpResponse, Transport, TransportEvent, TransportFactory, + }; + + struct NoNetwork; + #[async_trait::async_trait] + impl TransportFactory for NoNetwork { + async fn create_transport( + &self, + ) -> anyhow::Result<(Arc, async_channel::Receiver)> { + panic!("a played receipt probe must never connect") + } + } + #[async_trait::async_trait] + impl HttpClient for NoNetwork { + async fn execute(&self, _: HttpRequest) -> anyhow::Result { + panic!("a played receipt probe must never use HTTP") + } + } + + // Observe the real core builder before the deliberately absent socket rejects + // transmission. This proves wire construction, not server/device delivery. + #[tokio::test] + async fn core_played_wire_matches_preserved_bundle_privacy_and_addressing() { + for (index, (chat, disabled, expected, participant)) in [ + ("559900000001@s.whatsapp.net", false, "played", false), + ("111000011112222@lid", false, "played", false), + ("111000011112222@lid", true, "played-self", false), + ("120363000000000001@g.us", true, "played", true), + ("120363000000000001@g.us", false, "played", true), + ("120363000000000002@newsletter", false, "played-self", false), + ] + .into_iter() + .enumerate() + { + let db = SqliteStore::open(&format!( + "file:played_wire_{}_{}?mode=memory&cache=shared", + std::process::id(), + index + )) + .await + .unwrap(); + let pm = Arc::new(PersistenceManager::new(Arc::new(db)).await.unwrap()); + pm.modify_device(|device| device.read_receipts_disabled = disabled) + .await; + let (client, _events) = Client::builder() + .with_runtime(whatsapp_rust::TokioRuntime) + .with_persistence_manager(pm) + .with_transport_factory(NoNetwork) + .with_http_client(NoNetwork) + .build() + .await + .unwrap() + .into_parts(); + let waiter = client.wait_for_sent_node(NodeFilter::tag("receipt")); + let chat: Jid = chat.parse().unwrap(); + let sender: Jid = "111000011113333@lid".parse().unwrap(); + assert!( + send_played( + &client, + &chat, + participant.then_some(&sender), + "SYNTHETIC-AUDIO" + ) + .await + .is_err() + ); + let node = tokio::time::timeout(std::time::Duration::from_secs(1), waiter) + .await + .expect("core constructed the receipt before rejecting the absent socket") + .unwrap(); + let attr = |name| node.attrs.get(name).map(|v| v.as_str().to_string()); + assert_eq!(attr("type").as_deref(), Some(expected)); + assert_eq!(attr("to"), Some(chat.to_string())); + assert_eq!(attr("id").as_deref(), Some("SYNTHETIC-AUDIO")); + assert_eq!(attr("participant"), participant.then(|| sender.to_string())); + assert!(attr("t").is_some()); + } + } +}