torttube/sidecar/crates/torttube-sidecar/src/resolve.rs

// resolve.rs — Tier 1 (rustypipe) → Tier 2 (yt-dlp -j fallback)
// SPDX-License-Identifier: GPL-3.0-or-later

use serde_json::Value;

use crate::{run_yt_dlp, HandlerError};

/// Search YouTube via rustypipe. Returns `SearchResult<VideoItem>` as JSON —
/// the Python addon picks the fields it needs (id, name, channel, duration,
/// thumbnail, view_count, …) to build a Kodi directory listing.
pub(crate) async fn search(query: &str, limit: u32) -> Result<Value, HandlerError> {
    use rustypipe::client::RustyPipe;
    use rustypipe::model::VideoItem;

    let rp = RustyPipe::new();
    let result = rp
        .query()
        .search::<VideoItem, _>(query)
        .await
        .map_err(|e| classify_rustypipe_error(&e))?;

    // SearchResult<VideoItem>.items is a Paginator<VideoItem> — take the first
    // page truncated to `limit` items.
    let items_json: Vec<Value> = result
        .items
        .items
        .iter()
        .take(limit as usize)
        .filter_map(|v| serde_json::to_value(v).ok())
        .collect();

    tracing::info!(query, count = items_json.len(), "search ok via rustypipe");

    Ok(serde_json::json!({
        "source": "rustypipe",
        "query": query,
        "items": items_json,
        "corrected_query": result.corrected_query,
    }))
}

/// List a channel's recent videos. Returns the same VideoItem shape as
/// `search`, plus channel metadata (name, subscribers, description, banner).
pub(crate) async fn channel_videos(channel_id: &str, limit: u32) -> Result<Value, HandlerError> {
    use rustypipe::client::RustyPipe;

    let rp = RustyPipe::new();
    let ch = rp
        .query()
        .channel_videos(channel_id)
        .await
        .map_err(|e| classify_rustypipe_error(&e))?;

    let items_json: Vec<Value> = ch
        .content
        .items
        .iter()
        .take(limit as usize)
        .filter_map(|v| serde_json::to_value(v).ok())
        .collect();

    tracing::info!(channel_id, count = items_json.len(), "channel_videos ok");

    Ok(serde_json::json!({
        "source": "rustypipe",
        "channel": {
            "id": ch.id,
            "name": ch.name,
            "description": ch.description,
            "subscribers": ch.subscriber_count,
            "video_count": ch.video_count,
            "avatar": ch.avatar,
            "banner": ch.banner,
        },
        "items": items_json,
    }))
}

/// Fetch a subscriptions feed: pull the most recent N videos from each
/// channel in parallel, merge + sort by publish-date newest-first, cap
/// the returned union at `limit`. A channel that fails (404, region block,
/// rustypipe error) is silently dropped — one dead subscription shouldn't
/// kill the whole feed.
pub(crate) async fn subscriptions_feed(
    channel_ids: &[String],
    per_channel: u32,
    limit: u32,
) -> Result<Value, HandlerError> {
    use rustypipe::client::RustyPipe;
    use std::collections::HashMap;

    let rp = std::sync::Arc::new(RustyPipe::new());

    // Spawn one tokio task per channel; merge results once they all finish.
    // Fan-out is capped at 200 by main.rs; per_channel capped at 50.
    let tasks: Vec<_> = channel_ids
        .iter()
        .map(|cid| {
            let rp = rp.clone();
            let cid = cid.clone();
            tokio::spawn(async move {
                let res = rp.query().channel_videos(&cid).await;
                (cid, res)
            })
        })
        .collect();

    let mut all_items: Vec<Value> = Vec::new();
    let mut channel_names: HashMap<String, String> = HashMap::new();
    let mut failed: Vec<String> = Vec::new();

    for handle in tasks {
        let (cid, res) = match handle.await {
            Ok(t) => t,
            Err(e) => {
                tracing::warn!(error = %e, "subscriptions feed task panicked");
                continue;
            }
        };
        match res {
            Ok(ch) => {
                if !ch.name.is_empty() {
                    channel_names.insert(cid.clone(), ch.name.clone());
                }
                for vi in ch.content.items.iter().take(per_channel as usize) {
                    if let Ok(v) = serde_json::to_value(vi) {
                        all_items.push(v);
                    }
                }
            }
            Err(e) => {
                tracing::warn!(channel = %cid, error = %e, "subscriptions feed channel failed");
                failed.push(cid);
            }
        }
    }

    // Sort newest-first by publish_date. rustypipe VideoItem has an optional
    // `publish_date` as an RFC3339-ish string when known; fall back to
    // `publish_date_txt` (relative) ordering by string is meaningless, so we
    // keep channel-order as a stable secondary sort by leaving channel order
    // as insertion order and using a stable sort.
    all_items.sort_by(|a, b| {
        let ad = a.get("publish_date").and_then(Value::as_str).unwrap_or("");
        let bd = b.get("publish_date").and_then(Value::as_str).unwrap_or("");
        bd.cmp(ad) // descending
    });
    all_items.truncate(limit as usize);

    tracing::info!(
        channels = channel_ids.len(),
        items = all_items.len(),
        failed = failed.len(),
        "subscriptions_feed ok"
    );

    Ok(serde_json::json!({
        "source": "rustypipe",
        "items": all_items,
        "channels_total": channel_ids.len(),
        "channels_failed": failed,
        "channel_names": channel_names,
    }))
}

/// List a playlist's videos. Returns the same VideoItem shape as search/channel.
pub(crate) async fn playlist(playlist_id: &str, limit: u32) -> Result<Value, HandlerError> {
    use rustypipe::client::RustyPipe;

    let rp = RustyPipe::new();
    let pl = rp
        .query()
        .playlist(playlist_id)
        .await
        .map_err(|e| classify_rustypipe_error(&e))?;

    let items_json: Vec<Value> = pl
        .videos
        .items
        .iter()
        .take(limit as usize)
        .filter_map(|v| serde_json::to_value(v).ok())
        .collect();

    tracing::info!(playlist_id, count = items_json.len(), "playlist ok");

    Ok(serde_json::json!({
        "source": "rustypipe",
        "playlist": {
            "id": pl.id,
            "name": pl.name,
            "description": pl.description,
            "video_count": pl.video_count,
            "channel": pl.channel,
            "thumbnail": pl.thumbnail,
        },
        "items": items_json,
    }))
}

/// DASH-ready resolve: returns rustypipe's full `video_only_streams` +
/// `audio_streams` arrays + `details`. The Python addon builds an MPD
/// from these and hands it to inputstream.adaptive — unlocks 1080p+ via
/// H.264 hardware decode on the RPi (vs the 360p ceiling on progressive).
pub(crate) async fn resolve_dash(id: &str) -> Result<Value, HandlerError> {
    use rustypipe::client::RustyPipe;

    let rp = RustyPipe::new();
    let player = rp
        .query()
        .player(id)
        .await
        .map_err(|e| classify_rustypipe_error(&e))?;

    let details_json = serde_json::to_value(&player.details)
        .map_err(|e| HandlerError::Internal(format!("serialize details: {e}")))?;
    let video_streams = serde_json::to_value(&player.video_only_streams)
        .map_err(|e| HandlerError::Internal(format!("serialize video_only_streams: {e}")))?;
    let audio_streams = serde_json::to_value(&player.audio_streams)
        .map_err(|e| HandlerError::Internal(format!("serialize audio_streams: {e}")))?;

    tracing::info!(id, "resolve_dash ok via rustypipe");

    Ok(serde_json::json!({
        "source": "rustypipe",
        "details": details_json,
        "video_only_streams": video_streams,
        "audio_streams": audio_streams,
        "expires_in_seconds": player.expires_in_seconds,
    }))
}

/// Playback-ready single-URL resolve. Asks yt-dlp for `best[ext=mp4]/best` —
/// a combined audio+video format that Kodi can play as a plain HTTP URL.
/// Slower than `resolve()` (~3-5s) but guarantees a working stream.
pub(crate) async fn resolve_play(id: &str) -> Result<Value, HandlerError> {
    let url = format!("https://www.youtube.com/watch?v={id}");
    // -f best[ext=mp4]/best — prefer mp4 progressive, else any best combined.
    // We use -j to get the full info dump; the selected format's URL appears
    // as the top-level "url" field.
    let stdout = run_yt_dlp(&[
        "-j", "--no-warnings", "--no-playlist",
        "-f", "best[ext=mp4]/best",
        &url,
    ])
    .await
    .map_err(|e| classify_yt_dlp_error(&e))?;

    let dump: Value = serde_json::from_slice(&stdout)
        .map_err(|e| HandlerError::Extractor(format!("yt-dlp json parse: {e}")))?;

    let stream_url = dump
        .get("url")
        .and_then(Value::as_str)
        .ok_or_else(|| HandlerError::Extractor("yt-dlp: no top-level url".into()))?
        .to_string();

    tracing::info!(id, "resolve_play ok via yt-dlp combined");

    Ok(serde_json::json!({
        "source": "yt-dlp",
        "stream_url": stream_url,
        "title": dump.get("title"),
        "duration_s": dump.get("duration"),
        "channel_name": dump.get("channel"),
        "channel_id": dump.get("channel_id"),
        "thumbnail": dump.get("thumbnail"),
        "format_id": dump.get("format_id"),
        "ext": dump.get("ext"),
    }))
}

/// Metadata-rich resolve. Tries Tier 1 (rustypipe), falls back to Tier 2 (yt-dlp -j).
/// Returns the full extractor response including separate audio + video streams.
/// Use `resolve_play` instead for direct playback (returns a single combined URL).
pub(crate) async fn resolve(id: &str) -> Result<Value, HandlerError> {
    match tier1_rustypipe(id).await {
        Ok(v) => {
            tracing::info!(id, source = "rustypipe", "resolve ok");
            Ok(v)
        }
        Err(e) => {
            tracing::warn!(id, error = %e, "rustypipe failed; falling back to yt-dlp");
            // Typed errors that mean "video can't be played by anyone" — don't retry yt-dlp,
            // it'll just hit the same wall.
            if matches!(
                e,
                HandlerError::AgeRestricted
                    | HandlerError::PrivateVideo
                    | HandlerError::NotFound
            ) {
                return Err(e);
            }
            tier2_yt_dlp(id).await
        }
    }
}

/// Tier 1 — native rustypipe. Serializes the whole player.details + selected streams as
/// opaque pass-through JSON. The Python addon parses the fields it needs; this keeps us
/// resilient to rustypipe shape evolution and unblocks tier-2 normalization later.
async fn tier1_rustypipe(id: &str) -> Result<Value, HandlerError> {
    use rustypipe::client::RustyPipe;
    use rustypipe::param::StreamFilter;

    let rp = RustyPipe::new();
    let player = rp
        .query()
        .player(id)
        .await
        .map_err(|e| classify_rustypipe_error(&e))?;

    let (video, audio) = player.select_video_audio_stream(&StreamFilter::default());

    let details_json = serde_json::to_value(&player.details)
        .map_err(|e| HandlerError::Internal(format!("serialize details: {e}")))?;
    let video_json = video
        .map(|v| serde_json::to_value(v))
        .transpose()
        .map_err(|e| HandlerError::Internal(format!("serialize video: {e}")))?
        .unwrap_or(Value::Null);
    let audio_json = audio
        .map(|a| serde_json::to_value(a))
        .transpose()
        .map_err(|e| HandlerError::Internal(format!("serialize audio: {e}")))?
        .unwrap_or(Value::Null);

    Ok(serde_json::json!({
        "source": "rustypipe",
        "details": details_json,
        "video_stream": video_json,
        "audio_stream": audio_json,
    }))
}

/// Classify a yt-dlp shell-out error into one of our typed handler errors.
/// yt-dlp's stderr is freeform English; we match on **word-boundary** patterns
/// so "private" matches the standalone word, not e.g. "private network" inside
/// a TLS error. Preserve the original message verbatim in the returned error.
fn classify_yt_dlp_error(e: &anyhow::Error) -> HandlerError {
    let original = e.to_string();
    let lower = original.to_lowercase();
    if matches_word(&lower, &["age-restrict", "age restricted", "age restriction"]) {
        HandlerError::AgeRestricted
    } else if matches_word(&lower, &["private video", "video is private"]) {
        HandlerError::PrivateVideo
    } else if lower.contains("not available") || lower.contains("does not exist") {
        HandlerError::NotFound
    } else if matches_word(&lower, &["geo-restrict", "geo restrict", "region-restrict", "region restrict"]) {
        HandlerError::RegionBlocked
    } else {
        HandlerError::Extractor(original)
    }
}

/// Classify a rustypipe error into one of our typed handler errors.
/// rustypipe's error enum varies by version; we match on the Display string for resilience.
fn classify_rustypipe_error(e: &dyn std::fmt::Display) -> HandlerError {
    let original = e.to_string();
    let msg = original.to_lowercase();
    if matches_word(&msg, &["age-restrict", "age restricted", "age restriction"]) {
        HandlerError::AgeRestricted
    } else if matches_word(&msg, &["region-restrict", "region restrict", "geo-restrict", "geo restrict", "country restrict"]) {
        HandlerError::RegionBlocked
    } else if matches_word(&msg, &["private video", "video is private"]) {
        HandlerError::PrivateVideo
    } else if matches_word(&msg, &["not found", "unavailable", "does not exist"]) {
        HandlerError::NotFound
    } else if matches_word(&msg, &["network error", "timeout", "connection refused", "dns error"]) {
        HandlerError::Network(original)
    } else {
        HandlerError::Extractor(original)
    }
}

fn matches_word(haystack: &str, needles: &[&str]) -> bool {
    needles.iter().any(|n| haystack.contains(n))
}

/// Tier 2 — shell out to yt-dlp -j.
async fn tier2_yt_dlp(id: &str) -> Result<Value, HandlerError> {
    let url = format!("https://www.youtube.com/watch?v={id}");
    let stdout = run_yt_dlp(&["-j", "--no-warnings", "--no-playlist", &url])
        .await
        .map_err(|e| classify_yt_dlp_error(&e))?;

    let dump: Value = serde_json::from_slice(&stdout)
        .map_err(|e| HandlerError::Extractor(format!("yt-dlp json parse: {e}")))?;

    // yt-dlp's JSON has a `formats` array. We pass it through largely as-is — the addon
    // can pick what inputstream.adaptive wants. Shape it to match our protocol.
    let streams: Vec<Value> = dump
        .get("formats")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default()
        .into_iter()
        .filter_map(|f| {
            let url = f.get("url")?.as_str()?.to_string();
            let vcodec = f.get("vcodec").and_then(Value::as_str).unwrap_or("none");
            let acodec = f.get("acodec").and_then(Value::as_str).unwrap_or("none");
            let is_audio_only = vcodec == "none" && acodec != "none";
            let is_video_only = vcodec != "none" && acodec == "none";
            Some(serde_json::json!({
                "url": url,
                "itag": f.get("format_id").and_then(|v| v.as_str()).and_then(|s| s.parse::<u32>().ok()),
                "mime": f.get("ext"),
                "width": f.get("width"),
                "height": f.get("height"),
                "bitrate": f.get("tbr"),
                "is_audio_only": is_audio_only,
                "is_video_only": is_video_only,
            }))
        })
        .collect();

    Ok(serde_json::json!({
        "source": "yt-dlp",
        "title": dump.get("title"),
        "duration_s": dump.get("duration"),
        "channel_name": dump.get("channel"),
        "channel_id": dump.get("channel_id"),
        "streams": streams,
    }))
}