jellytau/src-tauri/src/player/stream_end.rs

//! Telling a *finished* stream apart from a *truncated* one.
//!
//! TRACES: UR-040 | DR-129 | UT-117
//!
//! Background audio-only playback of a video item streams a **progressive mp3
//! transcode over plain HTTP** (see
//! `OnlineRepository::build_audio_only_stream_url_for_video`). That response has
//! no reliable length — a live transcode is chunked — so when the connection
//! drops mid-episode the data source simply sees end-of-input. ExoPlayer cannot
//! distinguish that from the real end of the media and reports
//! `Player.STATE_ENDED`, which the app then treats as "the episode finished".
//!
//! The user-visible damage is not the missed advance itself. Playback parks in
//! ExoPlayer's `STATE_ENDED`, and the next play intent from the lockscreen,
//! notification or a Bluetooth reconnect goes through media3's
//! `Util.handlePlayButtonAction`, which seeks an ENDED player to its default
//! position before playing — so **the episode starts over from 0:00**. On a
//! flaky connection that reads as "it randomly restarts the episode".
//!
//! The player itself has no way to know; the *duration* does. Jellyfin gives us
//! the item's real runtime, so an end reported well short of it is a truncation,
//! not a finish — and the right response is to re-open the stream where it died,
//! which is the "buffer and resume" the user expects.

/// How far short of the item's runtime a stream may end and still count as a
/// natural finish.
///
/// Sized to swallow the two sources of slack in the comparison — the position
/// poll is up to 250 ms stale, and Jellyfin's reported runtime can disagree with
/// the transcoded output by a second or two — while staying far below the
/// minutes-long gap a dropped connection leaves. Erring long is the safe
/// direction: a false "finished" is the bug we are fixing, whereas a false
/// "truncated" only re-opens the stream for its last few seconds and then ends
/// again normally.
pub const TRUNCATED_STREAM_TOLERANCE_SECS: f64 = 10.0;

/// Consecutive resume attempts allowed at the same position before giving up.
///
/// A resume re-opens the same URL, so a server that is genuinely gone would
/// otherwise end → resume → end forever. Progress past the last attempt resets
/// the budget (see [`ResumeTracker`]), so this only bounds *stuck* retries.
pub const MAX_STALLED_RESUME_ATTEMPTS: u32 = 3;

/// Position change that counts as "this is a different playback context" —
/// either the resume made progress, or a different item is loaded.
const RESUME_PROGRESS_EPSILON_SECS: f64 = 1.0;

/// Did this end-of-stream happen far enough short of the item's runtime to be a
/// truncation rather than a finish?
///
/// `position` and `duration` must be on the same timeline — for a handoff stream
/// built with `StartTimeTicks`, that means the *absolute* position (handoff base
/// + the player's relative position) against the item's full runtime.
///
/// An unknown or non-positive `duration` answers `false`: with nothing to
/// compare against, the reported end is taken at face value (previous behaviour).
pub fn is_truncated_end(position: f64, duration: Option<f64>, tolerance: f64) -> bool {
    let Some(duration) = duration else {
        return false;
    };
    if duration <= 0.0 {
        return false;
    }
    position.max(0.0) + tolerance < duration
}

/// Rewrite an audio-only stream URL to start at `position_seconds`.
///
/// Resuming re-opens *the stream we were already playing*, so the URL is edited
/// in place rather than rebuilt from the repository: every other parameter —
/// `AudioStreamIndex` (the track the user picked in the video player),
/// `MediaSourceId`, `api_key` — is carried over untouched, and no network call
/// is needed to recover from a network failure.
pub fn with_start_time(url: &str, position_seconds: f64) -> String {
    let ticks = (position_seconds.max(0.0) * 10_000_000.0) as i64;
    let param = format!("StartTimeTicks={}", ticks);

    let (base, query) = match url.split_once('?') {
        Some((base, query)) => (base, query),
        // No query string at all: the URL was not built by us, but appending the
        // parameter is still the correct request to make.
        None => return format!("{}?{}", url, param),
    };

    let mut replaced = false;
    let mut parts: Vec<String> = query
        .split('&')
        .map(|part| {
            if part.split('=').next() == Some("StartTimeTicks") {
                replaced = true;
                param.clone()
            } else {
                part.to_string()
            }
        })
        .collect();

    if !replaced {
        parts.push(param);
    }

    format!("{}?{}", base, parts.join("&"))
}

/// The last playback time actually observed while media was loaded.
///
/// Some backends expose position and duration as **live** properties of the
/// loaded file — MPV's `time-pos` and `duration` stop resolving the moment it
/// unloads the file at EOF. Reading them straight through means that at exactly
/// the moment end-of-file handling wants to know where playback got to, the
/// answer is `0.0` / unknown: the player appears to rewind to 0:00 as it ends.
///
/// The polling thread records here, and the accessors fall back to it, so an EOF
/// reads as the last timestamp rather than as zero.
#[derive(Debug, Default, Clone, Copy)]
pub struct ObservedTime {
    position: f64,
    duration: Option<f64>,
}

impl ObservedTime {
    /// Record a live reading. Non-positive durations are treated as unknown —
    /// that is how a backend reports "not established yet", not a real zero.
    pub fn record(&mut self, position: f64, duration: f64) {
        self.position = position.max(0.0);
        if duration > 0.0 {
            self.duration = Some(duration);
        }
    }

    /// Record a position alone, e.g. straight after a seek, before the next poll.
    pub fn record_position(&mut self, position: f64) {
        self.position = position.max(0.0);
    }

    /// Forget everything — a different file is loading, and the previous one's
    /// timestamp must not leak into it.
    pub fn reset(&mut self) {
        *self = Self::default();
    }

    /// The live reading if there is one, else the last observed value.
    pub fn position_or_last(&self, live: Option<f64>) -> f64 {
        live.filter(|p| *p >= 0.0).unwrap_or(self.position)
    }

    /// The last observed position, with no live reading to prefer — the case
    /// where the *reporter* is the only source there is (webview-rendered media,
    /// which the native backend cannot see at all).
    pub fn last_position(&self) -> f64 {
        self.position
    }

    /// The last observed duration, if one was ever established.
    pub fn last_duration(&self) -> Option<f64> {
        self.duration
    }

    /// The live reading if there is one, else the last observed value.
    pub fn duration_or_last(&self, live: Option<f64>) -> Option<f64> {
        live.filter(|d| *d > 0.0).or(self.duration)
    }
}

/// Budget for consecutive resume attempts that make no progress.
///
/// Held by the player controller across ends of the *same* stream. Any position
/// change larger than [`RESUME_PROGRESS_EPSILON_SECS`] — the resume played on,
/// or a different item was loaded — is a fresh context and refills the budget.
#[derive(Debug, Default)]
pub struct ResumeTracker {
    last_position: Option<f64>,
    attempts: u32,
}

impl ResumeTracker {
    /// Record an attempt at `position`, returning its 1-based number — or `None`
    /// once the budget is spent. Callers use the number to back off: a stream
    /// that failed twice at the same spot is waiting on something slower than an
    /// immediate retry can outrun.
    pub fn allow_attempt(&mut self, position: f64) -> Option<u32> {
        let progressed = match self.last_position {
            Some(last) => (position - last).abs() > RESUME_PROGRESS_EPSILON_SECS,
            None => true,
        };
        if progressed {
            self.attempts = 0;
        }
        self.last_position = Some(position);
        self.attempts += 1;
        (self.attempts <= MAX_STALLED_RESUME_ATTEMPTS).then_some(self.attempts)
    }

    /// Forget the budget — a new item is playing, so nothing is stuck.
    pub fn reset(&mut self) {
        self.last_position = None;
        self.attempts = 0;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_end_near_duration_is_a_natural_finish() {
        // Episode runtime 25:00, stream ended at 24:56 — that is the end.
        assert!(!is_truncated_end(
            1496.0,
            Some(1500.0),
            TRUNCATED_STREAM_TOLERANCE_SECS
        ));
    }

    #[test]
    fn test_end_far_short_of_duration_is_truncated() {
        // Episode runtime 25:00, stream died at 10:00 — the connection dropped.
        assert!(is_truncated_end(
            600.0,
            Some(1500.0),
            TRUNCATED_STREAM_TOLERANCE_SECS
        ));
    }

    #[test]
    fn test_unknown_duration_is_taken_at_face_value() {
        // Nothing to compare against: keep the previous end-of-track behaviour
        // rather than resuming a stream that may really have finished.
        assert!(!is_truncated_end(
            600.0,
            None,
            TRUNCATED_STREAM_TOLERANCE_SECS
        ));
        assert!(!is_truncated_end(
            600.0,
            Some(0.0),
            TRUNCATED_STREAM_TOLERANCE_SECS
        ));
    }

    #[test]
    fn test_tolerance_boundary() {
        // Exactly one tolerance short still counts as finished, so poll staleness
        // and runtime rounding never fabricate a truncation.
        assert!(!is_truncated_end(1490.0, Some(1500.0), 10.0));
        assert!(is_truncated_end(1489.0, Some(1500.0), 10.0));
    }

    #[test]
    fn test_with_start_time_replaces_existing_ticks() {
        let url = "http://s/Audio/ep2/universal?api_key=k&AudioStreamIndex=2&StartTimeTicks=1200000000&Container=mp3";
        let out = with_start_time(url, 600.0);
        assert_eq!(
            out,
            "http://s/Audio/ep2/universal?api_key=k&AudioStreamIndex=2&StartTimeTicks=6000000000&Container=mp3"
        );
    }

    #[test]
    fn test_with_start_time_appends_when_absent() {
        // The next-episode stream is built without StartTimeTicks.
        let url = "http://s/Audio/ep3/universal?api_key=k&AudioStreamIndex=0";
        let out = with_start_time(url, 90.0);
        assert_eq!(
            out,
            "http://s/Audio/ep3/universal?api_key=k&AudioStreamIndex=0&StartTimeTicks=900000000"
        );
    }

    #[test]
    fn test_with_start_time_preserves_selected_audio_track() {
        // The whole point of editing the URL instead of rebuilding it: the track
        // the user chose in the video player survives the resume.
        let url = "http://s/Audio/ep2/universal?AudioStreamIndex=3&MediaSourceId=src-1";
        let out = with_start_time(url, 10.0);
        assert!(out.contains("AudioStreamIndex=3"));
        assert!(out.contains("MediaSourceId=src-1"));
    }

    #[test]
    fn test_with_start_time_without_query() {
        assert_eq!(
            with_start_time("http://s/Audio/ep2/universal", 1.0),
            "http://s/Audio/ep2/universal?StartTimeTicks=10000000"
        );
    }

    /// The bug: MPV unloads the file at EOF, so `time-pos` stops resolving and a
    /// straight read reports 0.0 — the position collapses to zero at precisely
    /// the moment end-of-file handling needs to know where playback reached.
    #[test]
    fn test_eof_reads_as_the_last_observed_timestamp() {
        let mut observed = ObservedTime::default();
        observed.record(178.0, 180.0);

        // The file is gone: both live properties fail.
        assert_eq!(observed.position_or_last(None), 178.0);
        assert_eq!(observed.duration_or_last(None), Some(180.0));
    }

    #[test]
    fn test_live_readings_win_while_the_file_is_loaded() {
        let mut observed = ObservedTime::default();
        observed.record(178.0, 180.0);

        assert_eq!(observed.position_or_last(Some(12.0)), 12.0);
        assert_eq!(observed.duration_or_last(Some(240.0)), Some(240.0));
    }

    #[test]
    fn test_unestablished_duration_is_not_recorded_as_zero() {
        let mut observed = ObservedTime::default();
        // A backend reports 0.0 for "duration not known yet", not a real zero.
        observed.record(5.0, 0.0);
        assert_eq!(observed.duration_or_last(None), None);
        assert_eq!(observed.position_or_last(None), 5.0);

        observed.record(6.0, 180.0);
        assert_eq!(observed.duration_or_last(Some(0.0)), Some(180.0));
    }

    #[test]
    fn test_reset_stops_the_previous_file_leaking_into_the_next() {
        let mut observed = ObservedTime::default();
        observed.record(178.0, 180.0);
        observed.reset();

        assert_eq!(observed.position_or_last(None), 0.0);
        assert_eq!(observed.duration_or_last(None), None);
    }

    #[test]
    fn test_seek_updates_the_last_position_before_the_next_poll() {
        let mut observed = ObservedTime::default();
        observed.record(10.0, 180.0);
        observed.record_position(120.0);

        assert_eq!(observed.position_or_last(None), 120.0);
        assert_eq!(
            observed.duration_or_last(None),
            Some(180.0),
            "seeking does not change how long the file is"
        );
    }

    #[test]
    fn test_resume_tracker_bounds_stalled_retries() {
        let mut tracker = ResumeTracker::default();
        // Same position over and over: the stream is not recovering.
        for n in 1..=MAX_STALLED_RESUME_ATTEMPTS {
            assert_eq!(
                tracker.allow_attempt(600.0),
                Some(n),
                "attempts are numbered so callers can back off"
            );
        }
        assert_eq!(
            tracker.allow_attempt(600.0),
            None,
            "a stream that ends at the same position every time must stop retrying"
        );
    }

    #[test]
    fn test_resume_tracker_refills_after_progress() {
        let mut tracker = ResumeTracker::default();
        for _ in 0..MAX_STALLED_RESUME_ATTEMPTS {
            tracker.allow_attempt(600.0);
        }
        assert_eq!(tracker.allow_attempt(600.0), None);
        // The next drop happened further in — the resumes are working, so the
        // budget must not be exhausted by earlier trouble.
        assert_eq!(tracker.allow_attempt(900.0), Some(1));
    }

    #[test]
    fn test_resume_tracker_reset() {
        let mut tracker = ResumeTracker::default();
        for _ in 0..=MAX_STALLED_RESUME_ATTEMPTS {
            tracker.allow_attempt(600.0);
        }
        tracker.reset();
        assert_eq!(tracker.allow_attempt(600.0), Some(1));
    }
}