youtube-legend-cli 0.4.0

//! The per-attempt ledger the error envelope publishes.
//!
//! One reason to change: what a chain attempt has to report about
//! itself, and how that report is spelled on the wire.

use serde::Serialize;
use std::sync::{Arc, Mutex};

use crate::error::AppError;

// Referenced only from the doc links below, so it is pulled in for the
// documentation build alone.
#[cfg(doc)]
use crate::provider::Provider;

/// What one provider attempt produced.
///
/// The spellings are fixed by the `outcome` enum of
/// `docs/schemas/error-envelope.schema.json`; adding a variant here
/// without adding it there breaks the published contract.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum AttemptOutcome {
    /// The provider answered with a usable body.
    Delivered,
    /// The provider reported that the video publishes no captions.
    NoCaptions,
    /// Every track the video publishes was machine-generated and none
    /// was served.
    AsrRefused,
    /// The video publishes captions, but not in the requested language.
    LanguageUnavailable,
    /// The provider could not answer, for any reason that is not one of
    /// the more specific values below.
    Unavailable,
    /// The upstream answered HTTP 429.
    RateLimited,
    /// The upstream served a challenge widget instead of an answer.
    Captcha,
    /// The attempt ran out of time before the page answered.
    DomTimeout,
    /// No Chromium or Chrome executable could be located.
    BrowserMissing,
    /// The chain did not call this provider because an earlier entry
    /// backed by the same upstream had already failed in this run.
    SkippedDegraded,
    /// The chain did not call this provider because it was turned off.
    SkippedDisabled,
}

/// One provider attempt, exactly as the error envelope publishes it.
///
/// Optional fields are omitted rather than serialised as `null`,
/// following the convention `commands::JsonError` already uses.
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct ProviderAttempt {
    /// Stable provider identifier, matching [`Provider::name`].
    pub provider: &'static str,
    /// What the attempt produced.
    pub outcome: AttemptOutcome,
    /// Wall-clock cost of the attempt. Absent for a provider the chain
    /// never called, because zero would claim a measurement.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub elapsed_ms: Option<u64>,
    /// Upstream HTTP status, when the failure carried one.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub http_status: Option<u16>,
    /// Body size in bytes, present on a delivered attempt.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub body_len: Option<usize>,
    /// Verbatim explanation the upstream produced, when it produced one.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub diagnostic: Option<String>,
}

/// Map a chain failure onto the `outcome` value the schema publishes.
pub(super) fn attempt_outcome(err: &AppError) -> AttemptOutcome {
    match err {
        AppError::NoSubtitle(crate::error::NoSubtitleReason::LanguageUnavailable)
        | AppError::LanguageUnavailable { .. } => AttemptOutcome::LanguageUnavailable,
        AppError::NoSubtitle(_) => AttemptOutcome::NoCaptions,
        AppError::CaptionsAsrOnly { .. } => AttemptOutcome::AsrRefused,
        AppError::RateLimited { .. } => AttemptOutcome::RateLimited,
        AppError::CaptchaChallenge { .. } => AttemptOutcome::Captcha,
        AppError::BrowserNotFound(_) => AttemptOutcome::BrowserMissing,
        // `dom_timeout` is the only timeout-shaped value the schema
        // publishes, and every provider that can run out of time here
        // drives a page. Folding this into `unavailable` would erase the
        // one fact the operator can act on: the deadline expired instead
        // of the upstream refusing.
        AppError::Timeout(_) => AttemptOutcome::DomTimeout,
        // Transport failures, bodies we do not model, and local defects
        // have no counterpart in the published set, so they report the
        // generic refusal rather than inventing a twelfth value.
        _ => AttemptOutcome::Unavailable,
    }
}

/// Longest explanation one attempt may carry into the envelope.
///
/// The provider call sites hand over a one-line message produced by an
/// in-page script, never a response body, a cookie or a header, so no
/// credential material reaches here; the cap exists so a pathological
/// message cannot inflate the envelope.
const MAX_DIAGNOSTIC_CHARS: usize = 200;

tokio::task_local! {
    /// Where a provider leaves its upstream's verbatim explanation for
    /// the attempt currently in flight.
    ///
    /// Task-local rather than a field of [`ProviderChain`]:
    /// `commands::batch` drives many concurrent fetches over ONE shared
    /// chain, so a slot on the chain would mix the explanations of
    /// different videos.
    pub(crate) static UPSTREAM_DIAGNOSTIC: Arc<Mutex<Option<String>>>;
}

tokio::task_local! {
    /// Where the chain mirrors every attempt it has finished, for a
    /// caller that may never receive the chain's own return value.
    ///
    /// `--timeout` wraps the whole fetch in `tokio::time::timeout`, and
    /// a fired deadline DROPS the future along with the `Vec` the walk
    /// was building. The caller then reported `attempts: 0`, which is
    /// indistinguishable from "nothing was tried" even after ninety
    /// seconds of real work. The `Arc` behind this key is created by
    /// the caller, OUTSIDE the cancelled future, so what the chain has
    /// already written survives the cancellation.
    pub(crate) static ATTEMPT_LEDGER: Arc<Mutex<Vec<ProviderAttempt>>>;
}

/// Append `attempt` to the walk's own ledger and mirror it to the
/// caller's cancellation-proof sink.
///
/// The mirror is a no-op outside [`ATTEMPT_LEDGER`], so a caller that
/// never set the key — `commands::batch`, and every test that drives a
/// chain directly — keeps exactly the behaviour it had.
pub(super) fn publish(local: &mut Vec<ProviderAttempt>, attempt: ProviderAttempt) {
    let _ = ATTEMPT_LEDGER.try_with(|sink| {
        sink.lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push(attempt.clone());
    });
    local.push(attempt);
}

/// Record `text` as the explanation of the attempt in flight.
///
/// A no-op outside a chain attempt, so a provider exercised directly by
/// a test never panics on the missing scope.
pub(crate) fn record_upstream_diagnostic(text: &str) {
    let truncated: String = text.chars().take(MAX_DIAGNOSTIC_CHARS).collect();
    let _ = UPSTREAM_DIAGNOSTIC.try_with(|slot| {
        *slot
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(truncated);
    });
}

#[cfg(test)]
mod attempt_outcome_tests {
    use super::*;
    use crate::error::AppResult;
    use crate::error::NoSubtitleReason;
    use crate::provider::{Format, Provider, ProviderChain, SubtitleInfo};
    use async_trait::async_trait;
    use std::time::Duration;

    /// Serialise an outcome the way the envelope does.
    fn wire(outcome: AttemptOutcome) -> String {
        serde_json::to_value(outcome)
            .expect("an outcome serialises")
            .as_str()
            .expect("an outcome is a string")
            .to_string()
    }

    /// Every value the schema publishes must serialise to the exact
    /// spelling the schema publishes. A rename here would emit an
    /// `outcome` no consumer can branch on.
    #[test]
    fn every_outcome_serialises_to_its_published_spelling() {
        for (outcome, published) in [
            (AttemptOutcome::Delivered, "delivered"),
            (AttemptOutcome::NoCaptions, "no_captions"),
            (AttemptOutcome::AsrRefused, "asr_refused"),
            (AttemptOutcome::LanguageUnavailable, "language_unavailable"),
            (AttemptOutcome::Unavailable, "unavailable"),
            (AttemptOutcome::RateLimited, "rate_limited"),
            (AttemptOutcome::Captcha, "captcha"),
            (AttemptOutcome::DomTimeout, "dom_timeout"),
            (AttemptOutcome::BrowserMissing, "browser_missing"),
            (AttemptOutcome::SkippedDegraded, "skipped_degraded"),
            (AttemptOutcome::SkippedDisabled, "skipped_disabled"),
        ] {
            assert_eq!(wire(outcome), published, "{outcome:?}");
        }
    }

    /// The mapping, over every variant the chain can hand it.
    ///
    /// The list is built from real `AppError` values rather than from a
    /// list of strings, because a list of strings would drift from the
    /// code exactly the way the schema once did.
    #[test]
    fn every_chain_error_maps_to_a_published_outcome() {
        let cases: Vec<(AppError, AttemptOutcome)> = vec![
            (
                AppError::NoSubtitle(NoSubtitleReason::NotPublished),
                AttemptOutcome::NoCaptions,
            ),
            (
                AppError::NoSubtitle(NoSubtitleReason::NotFound),
                AttemptOutcome::NoCaptions,
            ),
            (
                AppError::NoSubtitle(NoSubtitleReason::PrivateOrAgeRestricted),
                AttemptOutcome::NoCaptions,
            ),
            (
                AppError::NoSubtitle(NoSubtitleReason::Gone),
                AttemptOutcome::NoCaptions,
            ),
            (
                AppError::NoSubtitle(NoSubtitleReason::UnavailableForLegalReasons),
                AttemptOutcome::NoCaptions,
            ),
            // The one reason that is NOT an absence of captions.
            (
                AppError::NoSubtitle(NoSubtitleReason::LanguageUnavailable),
                AttemptOutcome::LanguageUnavailable,
            ),
            (
                AppError::LanguageUnavailable {
                    available: vec!["pt-BR".to_string()],
                },
                AttemptOutcome::LanguageUnavailable,
            ),
            (
                AppError::CaptionsAsrOnly {
                    asr_languages: vec!["pt".to_string()],
                },
                AttemptOutcome::AsrRefused,
            ),
            (
                AppError::ProviderUnavailable {
                    provider: "provider-decopy",
                },
                AttemptOutcome::Unavailable,
            ),
            (
                AppError::RateLimited {
                    provider: "provider-noiz",
                    retry_after_secs: Some(30),
                },
                AttemptOutcome::RateLimited,
            ),
            (
                AppError::CaptchaChallenge {
                    provider: "provider-decopy",
                    kind: "cf-turnstile",
                },
                AttemptOutcome::Captcha,
            ),
            (
                AppError::BrowserNotFound("chrome missing".to_string()),
                AttemptOutcome::BrowserMissing,
            ),
            (
                AppError::Timeout("after 30s".to_string()),
                AttemptOutcome::DomTimeout,
            ),
            // The population with no published counterpart: each one
            // reports the generic refusal instead of a twelfth value.
            (
                AppError::ProviderProtocolError {
                    provider: "provider-noiz",
                    detail: "missing field".to_string(),
                },
                AttemptOutcome::Unavailable,
            ),
            (
                AppError::TimedtextUpstreamError("unexpected EOF".to_string()),
                AttemptOutcome::Unavailable,
            ),
            (
                AppError::SubtitleTooLarge(60_000_000),
                AttemptOutcome::Unavailable,
            ),
            (
                AppError::Io(std::io::Error::other("disk gone")),
                AttemptOutcome::Unavailable,
            ),
            (
                AppError::Internal("invariant".to_string()),
                AttemptOutcome::Unavailable,
            ),
            (
                AppError::InvalidUsage("srt from a transcript".to_string()),
                AttemptOutcome::Unavailable,
            ),
        ];
        for (err, expected) in cases {
            assert_eq!(attempt_outcome(&err), expected, "{err:?}");
        }
    }

    /// The optional fields are omitted, never emitted as `null`: the
    /// schema forbids nothing here, but a consumer must be able to tell
    /// "not measured" from "measured as zero".
    #[test]
    fn an_attempt_omits_the_fields_it_did_not_measure() {
        let attempt = ProviderAttempt {
            provider: "provider-decopy",
            outcome: AttemptOutcome::SkippedDegraded,
            elapsed_ms: None,
            http_status: None,
            body_len: None,
            diagnostic: None,
        };
        let value = serde_json::to_value(&attempt).expect("serialises");
        let object = value.as_object().expect("an attempt is an object");
        assert_eq!(object.len(), 2, "only the required pair survives: {value}");
        assert_eq!(object["provider"], serde_json::json!("provider-decopy"));
        assert_eq!(object["outcome"], serde_json::json!("skipped_degraded"));
    }

    /// The diagnostic channel is a no-op outside a chain attempt, so a
    /// provider exercised directly by a test never panics on it.
    #[tokio::test]
    async fn recording_a_diagnostic_outside_an_attempt_is_a_no_op() {
        record_upstream_diagnostic("the page said no");
    }

    /// Inside an attempt the words survive, truncated to the published
    /// cap so a pathological message cannot inflate the envelope.
    #[tokio::test]
    async fn a_diagnostic_survives_the_attempt_and_is_capped() {
        let sink: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
        let long = "x".repeat(MAX_DIAGNOSTIC_CHARS + 50);
        UPSTREAM_DIAGNOSTIC
            .scope(Arc::clone(&sink), async {
                record_upstream_diagnostic(&long);
            })
            .await;
        let recorded = sink.lock().expect("uncontended").take().expect("recorded");
        assert_eq!(recorded.chars().count(), MAX_DIAGNOSTIC_CHARS);
    }

    /// The ledger has to survive a cancelled walk: `--timeout` drops
    /// the whole future, and reporting the empty `Vec` published
    /// `attempts: 0` for a run that had really tried — the same shape a
    /// run that tried nothing produces. The attempt still in flight
    /// when the deadline fires stays absent, because inventing it would
    /// claim a measurement nobody took.
    #[tokio::test]
    async fn the_finished_attempts_outlive_a_cancelled_walk() {
        struct Refusing;
        #[async_trait]
        impl Provider for Refusing {
            fn name(&self) -> &'static str {
                "mock-refusing"
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                Err(AppError::CaptchaChallenge {
                    provider: "mock-refusing",
                    kind: "cf-turnstile",
                })
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                unreachable!("fetch_subtitle already failed")
            }
        }

        /// Never answers, standing in for the upstream that burns the
        /// whole budget.
        struct Hanging;
        #[async_trait]
        impl Provider for Hanging {
            fn name(&self) -> &'static str {
                "mock-hanging"
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                std::future::pending().await
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                unreachable!("fetch_subtitle never returns")
            }
        }

        let chain = ProviderChain::with_min_interval(
            vec![Box::new(Refusing), Box::new(Hanging)],
            Duration::from_millis(1),
        );
        let sink: Arc<Mutex<Vec<ProviderAttempt>>> = Arc::new(Mutex::new(Vec::new()));
        let cancelled = tokio::time::timeout(
            Duration::from_millis(250),
            chain.fetch_subtitle_traced_into("dQw4w9WgXcQ", "en", Format::Srt, &sink),
        )
        .await;
        assert!(cancelled.is_err(), "the deadline has to fire");

        let recovered = sink.lock().expect("uncontended");
        assert_eq!(
            recovered.len(),
            1,
            "the finished attempt survives and the one in flight is not invented: {recovered:?}"
        );
        assert_eq!(recovered[0].provider, "mock-refusing");
        assert_eq!(recovered[0].outcome, AttemptOutcome::Captcha);
        assert!(
            recovered[0].elapsed_ms.is_some(),
            "a finished attempt carries its measurement: {recovered:?}"
        );
    }

    /// The other half of the same guarantee, and the one that keeps the
    /// fix honest: when the deadline fires while the FIRST attempt is
    /// still in flight, nothing has finished, so the sink stays empty
    /// and no entry is invented for the provider that never answered.
    ///
    /// Without this the first test could be satisfied by a chain that
    /// simply records an entry per provider it *reached*, which would
    /// publish a measurement nobody took.
    #[tokio::test]
    async fn no_attempt_is_invented_when_the_deadline_fires_mid_flight() {
        struct Hanging;
        #[async_trait]
        impl Provider for Hanging {
            fn name(&self) -> &'static str {
                "mock-hanging"
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                std::future::pending().await
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                unreachable!("fetch_subtitle never returns")
            }
        }

        let chain =
            ProviderChain::with_min_interval(vec![Box::new(Hanging)], Duration::from_millis(1));
        let sink: Arc<Mutex<Vec<ProviderAttempt>>> = Arc::new(Mutex::new(Vec::new()));
        let cancelled = tokio::time::timeout(
            Duration::from_millis(150),
            chain.fetch_subtitle_traced_into("dQw4w9WgXcQ", "en", Format::Srt, &sink),
        )
        .await;
        assert!(cancelled.is_err(), "the deadline has to fire");

        let recovered = sink.lock().expect("uncontended");
        assert!(
            recovered.is_empty(),
            "an attempt still in flight left no measurement, so it gets no entry: {recovered:?}"
        );
    }

    /// The whole stitch: a provider leaves its upstream's words in the
    /// channel, and the ledger the chain returns carries them — plus the
    /// entry for the provider that never ran, which is the fact nothing
    /// recorded before.
    ///
    /// `CaptchaChallenge` is the failure used here because it is
    /// degraded AND never retryable, so the walk costs no back-off.
    #[tokio::test]
    async fn the_ledger_carries_the_words_and_the_skip() {
        struct Refusing;
        #[async_trait]
        impl Provider for Refusing {
            fn name(&self) -> &'static str {
                "mock-refusing"
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                record_upstream_diagnostic("track list did not render within poll limit");
                Err(AppError::CaptchaChallenge {
                    provider: "mock-refusing",
                    kind: "cf-turnstile",
                })
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                unreachable!("fetch_subtitle already failed")
            }
        }

        let chain = ProviderChain::with_min_interval(
            vec![Box::new(Refusing), Box::new(Refusing)],
            Duration::from_millis(1),
        );
        let (result, attempts) = chain
            .fetch_subtitle_traced("dQw4w9WgXcQ", "en", Format::Srt)
            .await;
        assert!(result.is_err(), "both entries refuse");
        assert_eq!(attempts.len(), 2, "both entries are recorded: {attempts:?}");
        assert_eq!(attempts[0].provider, "mock-refusing");
        assert_eq!(attempts[0].outcome, AttemptOutcome::Captcha);
        assert_eq!(
            attempts[0].diagnostic.as_deref(),
            Some("track list did not render within poll limit")
        );
        assert!(attempts[0].elapsed_ms.is_some());
        assert_eq!(attempts[1].outcome, AttemptOutcome::SkippedDegraded);
        assert!(
            attempts[1].elapsed_ms.is_none(),
            "a provider that never ran has no measurement"
        );
    }
}