fix: track listened audio and send played receipts

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;

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
+    }
+}

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<Stream>,
+        started: Arc<PlaybackStart>,
         is_playing: Arc<AtomicBool>,
         position: Arc<AtomicUsize>,
         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::<i16>(
                     &device,
@@ -360,6 +370,7 @@ mod cpal_output {
                     is_playing,
                     finished,
                     completion_tx,
+                    self.started.clone(),
                 ),
                 SampleFormat::U16 => build_stream::<u16>(
                     &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<T: SizedSample + FromSample<f32>>(
         device: &cpal::Device,
         config: StreamConfig,
@@ -442,6 +456,7 @@ mod cpal_output {
         is_playing: Arc<AtomicBool>,
         finished: Arc<AtomicBool>,
         completion_tx: Arc<Mutex<Option<oneshot::Sender<()>>>>,
+        started: Arc<PlaybackStart>,
     ) -> Result<Stream, PlayerError> {
         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 {

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::<dyn FnMut()>::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;

crates/chat-store/migrations/2026-10-26-140000_audio_played/down.sql

@@ -0,0 +1 @@
+DROP TABLE audio_played_receipts;

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)
+);

crates/chat-store/migrations/2026-10-26-140001_pending_audio_plays/down.sql

@@ -0,0 +1 @@
+DROP TABLE audio_played_pending;

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)
+);

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<bool> {
+        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::<diesel::sql_types::Integer, _>(device_id)
+                        .bind::<diesel::sql_types::BigInt, _>(row.id)
+                        .get_result::<Count>(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<Option<StoredMessage>> {
         self.message_by_ownership(chat, msg_id, None).await
     }

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,
+    }
+}