crabidy/ytdy/src/ytdlp.rs

//! Minimal `yt-dlp` sidecar for **stream URLs only**
//! (see `architecture/youtube-rustypipe.md` D2-revised).
//!
//! All metadata comes from the pure-Rust rustypipe extractor; this
//! module exists because YouTube currently rejects sustained tokenless
//! stream fetching (every URL serves exactly its leading ~1 MiB), and
//! the token-capable Innertube clients need signature deciphering that
//! is broken in rustypipe upstream right now. `yt-dlp` solves the
//! cipher challenges and its URLs stream fully (throttled, but well
//! above audio bitrate). When upstream recovers, this sidecar can be
//! dropped without touching the provider: it is only consulted by
//! `get_urls_for_track`.
//!
//! Subprocess discipline: argv-only (never a shell), one bounded call
//! (`kill_on_drop`), stdout capped, stderr summarized to a short
//! suffix. Errors carry the video id and exit status — never stream
//! URLs (they may embed tokens).

use std::path::PathBuf;
use std::process::Stdio;
use std::time::Duration;

use tracing::debug;

/// Cap on captured stdout; `-g` prints a handful of URLs.
const MAX_STDOUT_BYTES: usize = 512 * 1024;

/// Errors from the sidecar.
#[derive(Debug, thiserror::Error)]
pub enum YtDlpError {
    #[error("cannot run yt-dlp: {0}")]
    Spawn(String),
    #[error("yt-dlp timed out after {0:?}")]
    Timeout(Duration),
    #[error("yt-dlp failed ({status}): {stderr}")]
    Failed { status: String, stderr: String },
    #[error("yt-dlp printed no stream url")]
    NoUrl,
}

/// A configured `yt-dlp` binary used for stream URL resolution.
#[derive(Clone, Debug)]
pub struct YtDlp {
    binary: PathBuf,
    timeout: Duration,
}

impl YtDlp {
    pub fn new(binary: PathBuf, timeout: Duration) -> Self {
        Self { binary, timeout }
    }

    /// Probes `--version`; used at init to decide whether the sidecar
    /// is available (absence degrades to rustypipe URLs, never fails
    /// the provider).
    pub async fn probe(&self) -> Result<String, YtDlpError> {
        let stdout = self.run(&["--version"]).await?;
        Ok(stdout.trim().to_string())
    }

    /// The audio stream URL for a watch URL: `audio/mp4` (AAC)
    /// preferred — the format the local player decodes.
    pub async fn stream_url(&self, watch_url: &str) -> Result<String, YtDlpError> {
        let stdout = self
            .run(&[
                "-f",
                "bestaudio[ext=m4a]/bestaudio/best",
                "-g",
                "--no-playlist",
                watch_url,
            ])
            .await?;
        stdout
            .lines()
            .map(str::trim)
            .find(|line| !line.is_empty())
            .map(str::to_string)
            .ok_or(YtDlpError::NoUrl)
    }

    /// Runs one bounded call and returns its stdout.
    async fn run(&self, args: &[&str]) -> Result<String, YtDlpError> {
        debug!(binary = %self.binary.display(), ?args, "running yt-dlp");
        let mut command = tokio::process::Command::new(&self.binary);
        command
            .args(["--no-warnings"])
            .args(args)
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        let child = command
            .spawn()
            .map_err(|err| YtDlpError::Spawn(err.to_string()))?;
        let output = tokio::time::timeout(self.timeout, child.wait_with_output())
            .await
            .map_err(|_| YtDlpError::Timeout(self.timeout))?
            .map_err(|err| YtDlpError::Spawn(err.to_string()))?;
        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr);
            let summary: String = stderr.trim().chars().take(200).collect();
            return Err(YtDlpError::Failed {
                status: output.status.to_string(),
                stderr: summary,
            });
        }
        let mut stdout = output.stdout;
        stdout.truncate(MAX_STDOUT_BYTES);
        Ok(String::from_utf8_lossy(&stdout).into_owned())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use std::os::unix::fs::PermissionsExt;
    use std::path::Path;

    /// A fake yt-dlp: a script dispatching on its argv.
    fn fake_binary(dir: &Path, body: &str) -> PathBuf {
        let path = dir.join("fake-yt-dlp");
        fs::write(&path, format!("#!/bin/sh\n{body}\n")).expect("write fake binary");
        let mut perms = fs::metadata(&path).expect("metadata").permissions();
        perms.set_mode(0o755);
        fs::set_permissions(&path, perms).expect("chmod");
        path
    }

    /// Retries a call while exec races a concurrent fork (ETXTBSY under
    /// parallel test load).
    async fn retry_spawn<T, F, Fut>(mut call: F) -> Result<T, YtDlpError>
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = Result<T, YtDlpError>>,
    {
        let mut last = call().await;
        for _ in 0..4 {
            if !matches!(last, Err(YtDlpError::Spawn(_))) {
                break;
            }
            last = call().await;
        }
        last
    }

    #[tokio::test]
    async fn resolves_stream_urls_and_probes() {
        let dir = tempfile::TempDir::new().expect("tempdir");
        let binary = fake_binary(
            dir.path(),
            r#"
case "$*" in
*--version*) echo "2026.07.04"; exit 0 ;;
*watch?v=vid1*) printf '%s\n' "https://example.test/a.m4a"; exit 0 ;;
*) echo boom >&2; exit 1 ;;
esac"#,
        );
        let ytdlp = YtDlp::new(binary, Duration::from_secs(30));
        assert_eq!(
            retry_spawn(|| ytdlp.probe()).await.expect("probe"),
            "2026.07.04"
        );
        let url = retry_spawn(|| ytdlp.stream_url("https://www.youtube.com/watch?v=vid1"))
            .await
            .expect("url");
        assert_eq!(url, "https://example.test/a.m4a");
        // Failures are typed and carry a bounded stderr summary.
        let err = ytdlp
            .stream_url("https://www.youtube.com/watch?v=nope")
            .await
            .expect_err("fails");
        assert!(err.to_string().contains("boom"), "{err}");
        // A missing binary is a typed spawn error.
        let gone = YtDlp::new(dir.path().join("gone"), Duration::from_secs(5));
        assert!(matches!(gone.probe().await, Err(YtDlpError::Spawn(_))));
    }

    #[tokio::test]
    async fn hung_calls_hit_the_timeout() {
        let dir = tempfile::TempDir::new().expect("tempdir");
        let binary = fake_binary(dir.path(), "sleep 30");
        let ytdlp = YtDlp::new(binary, Duration::from_millis(200));
        let started = std::time::Instant::now();
        let mut saw_timeout = false;
        for _ in 0..5 {
            match ytdlp.probe().await {
                Err(YtDlpError::Timeout(_)) => {
                    saw_timeout = true;
                    break;
                }
                // Under parallel test load, exec of a just-written
                // script can race a concurrent fork (ETXTBSY) — retry.
                Err(YtDlpError::Spawn(_)) => continue,
                other => panic!("expected a timeout, got {other:?}"),
            }
        }
        assert!(saw_timeout, "no timeout within 5 attempts");
        assert!(started.elapsed() < Duration::from_secs(10));
    }
}