meta-feeder-youtube/feeder-plugin/youtube-feeder/src/youtube.rs

//! The `youtube` upstream — the **delegated-playback** tier.
//!
//! Resolves a MusicBrainz release-group to per-track YouTube references so a
//! recording with no owned bytes anywhere is still *playable*.
//!
//! # Two tiers, and the line between them
//!
//! Everything in the "resolver" section below describes the **verified** paths
//! (`track`, `musicVideo`): id-anchored, channel- and duration-checked, and
//! built to refuse. A third path, [`YouTubePlugin::video_search`], answers a
//! plain `fileType:video` query with **unverified** free-text hits, because a
//! video client asking a video question deserves an answer even when nothing
//! can vouch for it.
//!
//! The two do not blur, and the reason is structural rather than editorial: raw
//! records never carry `anchored`, and a consumer's grid drops non-anchored
//! gateway hits unconditionally, so they can only ever land in a "raw sources"
//! section. Read the refusal doctrine below as scoped to the verified paths —
//! it is what keeps *those* trustworthy, and the raw tier is walled off from
//! them by the dispatch order in [`FeederPlugin::handle_query`], not by taste. It is the third
//! plugin in this binary, beside `musicbrainz` (identity) and
//! `internetarchive` (free bytes), and it lives here rather than in a repo of
//! its own because the MusicBrainz client, the rate budget, the cache and the
//! MBID→tracklist resolution it needs are all already written next door.
//!
//! # What makes this tier different from every other one
//!
//! Every other version in MetaMesh is *bytes at a cid meta-share can serve* —
//! owned already, or fetchable and then seeded. A YouTube reference is a third
//! thing: **playback we can only delegate**. Nothing is transported, seeded or
//! content-addressed, and the byte path REFUSES the cid (`400`) rather than
//! 404ing. See `docs/cid-formats.md` §7 and the delegated-playback study.
//!
//! Consequences that are architecture, not decoration:
//!
//! * The outcome is **metadata-only** — the `(None, Some)` branch, `bytes:
//!   None`, no `handle_fetch`, no `get_blob`, no bytes ever.
//! * `HashKind::YtVideo` keeps it out of the blockstore seed, which is gated on
//!   `Sha2_256` anyway — but the explicit kind is what lets the core, and any
//!   future consumer, tell "no bytes *yet*" from "no bytes, ever".
//! * A reference is an **on-ramp, not a replacement**: because it hangs off the
//!   same record as any future real cid, the version list upgrades in place and
//!   the tier ordering demotes the reference automatically.
//!
//! # The resolver, and why it refuses more than it emits
//!
//! The gateway path here is well-trodden; the real risk is **match quality**,
//! and the asymmetry that makes it matter: a *missing* track shows up as an
//! honestly sparse album page, while a *wrong* one shows up as a working app
//! playing the wrong song. So this resolver is built to refuse.
//!
//! Measured over 83 tracks / 12 albums, deliberately unkind sample (study §7.1):
//! 72.3% EXACT (channel **and** duration agree), 15.7% TITLE, 3.6% genuine coin
//! flips. **Only EXACT is emitted.** That is the deliberate trade: 72.3%
//! coverage and an honestly sparser page, rather than 88.0% with a confidence
//! marker the UI would have to render.
//!
//! The two filters are both load-bearing and neither is sufficient alone:
//!
//! * **Channel id == the artist's YouTube-Music channel** answers "the right
//!   artist, officially" as an exact id comparison rather than a bag of
//!   heuristics (`- Topic` suffixes, verified badges, string similarity).
//! * **Duration within ±10 s of the MusicBrainz recording length** answers "the
//!   right *recording*". MusicBrainz holds three "Get Lucky" recordings
//!   (237/294/40 s) and Piped returned 370 s and 249 s; without duration you
//!   cannot tell which one you queued.

use std::collections::{BTreeMap, HashMap};
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, GatewayQuery, HashKind, HashOutcome};
use meta_feeder_sdk::types::{ByteStream, DiscoveryRecord, GatewayError, Hash, PluginHealth};
use serde_json::Value;
use tracing::{debug, warn};

use meta_feeder_sdk::cache::MidhashCache;
use crate::consts::{
    MAX_REFS_PER_RECORDING, MB_BURST, MB_RATE_PER_SEC, MB_TIMEOUT_SECS, PIPED_INSTANCES,
    PIPED_TIMEOUT_SECS, USER_AGENT, VIDEO_RECORD_NS,
    YT_CANDIDATES_PER_TRACK, YT_DURATION_TOLERANCE_SECS, YT_THUMB_BASE, YT_VIDEO_SEARCH_MAX,
};
use musicbrainz_feeder::client::{primary_artist, MusicBrainzClient};
use meta_feeder_sdk::budget::RateBudget;

/// Which key produced a channel match, in the priority order the resolver
/// tries them.
///
/// **Recorded on the record**, not just used and dropped. Even emitting EXACT
/// only, the key that produced a match is the difference between "MusicBrainz
/// says this is the artist's official music channel" and "a name match against
/// YouTube's own artist index agreed" — and retrofitting the distinction onto
/// records already replicated across the mesh is painful.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ChannelKey {
    /// MusicBrainz `youtube music` relation — authoritative, sparse.
    MbYoutubeMusic,
    /// MusicBrainz `youtube` relation — the artist's *regular* channel.
    ///
    /// Weaker, and the weakness is specific: for BTS this is not where YouTube
    /// Music serves from, and matching on it scored **0/8** exact. It is not
    /// useless either — Nujabes and Boards of Canada both matched 8/8 through
    /// it, because artists with no separate YouTube Music artist channel get
    /// their regular uploads surfaced by `music_songs`. Second, never first.
    MbYoutube,
    /// Piped's own `music_artists` index — name-matched, so the weakest key,
    /// but it has **no coverage gap**, which is why it exists.
    ///
    /// MusicBrainz channel coverage looked like a hard ceiling (77% exact where
    /// a relation exists, **0%** where it does not — 0% *by construction*, since
    /// with no channel id nothing can reach EXACT however obviously right it
    /// is). Adding this third key lifted the sample from 53.0% to 72.3% EXACT
    /// and cut the risky residue from 4.8% to 3.6%.
    PipedMusicArtists,
}

impl ChannelKey {
    /// The value written to the record's `ytMatchKey` field.
    pub fn as_str(self) -> &'static str {
        match self {
            ChannelKey::MbYoutubeMusic => "mb:youtube-music",
            ChannelKey::MbYoutube => "mb:youtube",
            ChannelKey::PipedMusicArtists => "piped:music-artists",
        }
    }
}

/// An artist's candidate channels, best key first. Empty means "no key
/// available", which makes every track for that artist unresolvable — by
/// design, since EXACT requires a channel comparison.
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct ChannelKeys {
    /// `(channel_id, key_label)` pairs in priority order.
    pub ranked: Vec<(String, String)>,
}

/// `ytMatchKey` for the raw tier — the value that means "nothing verified this".
///
/// A *value*, not an absent field. Every record this feeder emits carries a
/// match key, so a reader never has to distinguish "unverified" from "written
/// by a build that predates match keys" — a distinction that is impossible to
/// recover once records are replicated across the mesh.
pub const RAW_MATCH_KEY: &str = "unverified:free-text";

/// One Piped search result, trimmed to what the filters read.
#[derive(Clone, Debug)]
struct Candidate {
    video_id: String,
    title: String,
    uploader_channel: String,
    duration_secs: i64,
    /// The channel's *display name*. The music paths never read it — they
    /// compare channel **ids**, which is the whole point of their filter — but
    /// the raw tier has no id to compare against and puts the uploader in the
    /// record's description instead, so a viewer can see who they are about to
    /// watch. Empty when Piped omits it.
    uploader: String,
}

/// The verdict for one (track, candidate) pair.
///
/// Only [`Verdict::Exact`] is ever emitted. The other variants exist so the
/// refusals are *named* — in particular [`Verdict::ChannelWrongCut`], which is
/// the single most dangerous shape in this tier and must never be widened into
/// a near-miss.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Verdict {
    /// Channel **and** duration agree. Safe to emit.
    Exact,
    /// The channel agrees but the runtime does not — a **refusal**, never a
    /// near-miss.
    ///
    /// # The artist-credit trap
    ///
    /// This is where classical breaks, and it breaks confidently. MusicBrainz
    /// credits the **composer**, not the performer:
    ///
    /// ```text
    /// Glenn Gould / The Goldberg Variations -> MB artist-credit = "Johann Sebastian Bach"
    /// ```
    ///
    /// So the resolver looks up *Bach's* channel, finds *a* Goldberg
    /// Variations on it, and the channel key — the strongest signal it has —
    /// agrees. The probe scored those at **+32 s and +209 s**: right composer,
    /// wrong performance. Widening the tolerance to "rescue" them would emit
    /// someone else's recording under this track's name.
    ChannelWrongCut,
    /// No channel agreement, or nothing plausible at all.
    NoMatch,
}

pub struct YouTubePlugin {
    http: reqwest::Client,
    cache: Option<MidhashCache>,
    mb: Option<Arc<MusicBrainzClient>>,
    /// Rotating cursor into [`PIPED_INSTANCES`] — see [`Self::piped_get`].
    instance: AtomicUsize,
    /// Test hook: pin Piped to a wiremock base instead of the pool.
    piped_base: Option<String>,
    /// Test hook: pin the MusicBrainz base.
    mb_base: Option<String>,
}

impl Default for YouTubePlugin {
    fn default() -> Self {
        Self {
            http: reqwest::Client::builder()
                .user_agent(USER_AGENT)
                .timeout(Duration::from_secs(PIPED_TIMEOUT_SECS))
                .build()
                .expect("build youtube http client"),
            cache: None,
            mb: None,
            instance: AtomicUsize::new(0),
            piped_base: None,
            mb_base: None,
        }
    }
}

impl YouTubePlugin {
    /// Test constructor: pin both upstreams at wiremock servers.
    pub fn with_bases(piped_base: String, mb_base: String) -> Self {
        Self {
            piped_base: Some(piped_base),
            mb_base: Some(mb_base),
            ..Default::default()
        }
    }

    /// One GET against the Piped pool, rotating past a dead instance.
    ///
    /// The rotation is the whole reliability story for this leg, and it is
    /// affordable precisely because we take only the 11-character video id: a
    /// dead instance costs a retry, not a broken stream.
    async fn piped_get(&self, path_and_query: &str) -> Option<Value> {
        if let Some(base) = &self.piped_base {
            let url = format!("{}/{}", base.trim_end_matches('/'), path_and_query);
            return self.http.get(&url).send().await.ok()?.json().await.ok();
        }
        let start = self.instance.load(Ordering::Relaxed);
        for i in 0..PIPED_INSTANCES.len() {
            let idx = (start + i) % PIPED_INSTANCES.len();
            let url = format!("{}/{}", PIPED_INSTANCES[idx], path_and_query);
            match self.http.get(&url).send().await {
                Ok(r) if r.status().is_success() => match r.json::<Value>().await {
                    Ok(v) => {
                        // Stick to a working instance for subsequent calls.
                        self.instance.store(idx, Ordering::Relaxed);
                        return Some(v);
                    }
                    Err(e) => debug!(target: "meta-music", %url, error = %e, "piped: bad json"),
                },
                Ok(r) => {
                    debug!(target: "meta-music", %url, status = %r.status(), "piped: non-2xx, rotating")
                }
                Err(e) => debug!(target: "meta-music", %url, error = %e, "piped: transport, rotating"),
            }
        }
        warn!(target: "meta-music", path = %path_and_query, "piped: every instance failed");
        None
    }

    /// Resolve an artist's channel keys, best first.
    ///
    /// Cached **positively only**: MusicBrainz answers `503` when crowded and
    /// the client reports that the same way it reports an empty result, so
    /// caching a negative would freeze a transport hiccup into "this artist has
    /// no channel" forever. The probe fabricated exactly that verdict — it
    /// reported "no YouTube channel in MusicBrainz" for artists that plainly
    /// have one — until the failure path was separated from the empty one.
    async fn channel_keys(&self, artist_mbid: &str, artist_name: &str) -> ChannelKeys {
        let cache_key = artist_mbid;
        if let Some(c) = self.cache.as_ref() {
            if let Ok(Some(hit)) = c.get_yt_channel(cache_key) {
                if let Ok(v) = serde_json::from_str::<ChannelKeys>(&hit) {
                    if !v.ranked.is_empty() {
                        return v;
                    }
                }
            }
        }

        let mut ranked: Vec<(String, String)> = Vec::new();
        let deadline = Duration::from_secs(MB_TIMEOUT_SECS);

        // Did the MusicBrainz leg actually *answer*? Not "did it find
        // anything" — see the caching rule at the bottom of this function.
        let mut mb_answered = artist_mbid.is_empty();

        if let Some(mb) = self.mb.as_ref() {
            if !artist_mbid.is_empty() {
                if let Some(rels) = mb.artist_url_relations(artist_mbid, deadline).await {
                    mb_answered = true;
                    // `youtube music` first, `youtube` second — the order is a
                    // measurement, not a preference. See `ChannelKey`.
                    for (want, key) in [
                        ("youtube music", ChannelKey::MbYoutubeMusic),
                        ("youtube", ChannelKey::MbYoutube),
                    ] {
                        for (ty, url) in &rels {
                            if ty.eq_ignore_ascii_case(want) {
                                if let Some(id) = channel_id_from_url(url) {
                                    ranked.push((id, key.as_str().to_string()));
                                }
                            }
                        }
                    }
                }
            }
        }

        // The third key. Tried even when MusicBrainz answered, because it costs
        // one search and covers the 4-of-12 albums MusicBrainz has no relation
        // for at all. Ordered last, so MusicBrainz still wins where it has an
        // answer — and where both answer they agree (Daft Punk resolves to
        // `UCRr1xG_2WIDs18a6cIiCxeA` either way).
        //
        // Tried under every name MusicBrainz knows, not just the credited one:
        // a non-Latin credit finds no exact-name channel even when the artist
        // plainly has one (§7.2.1, `植松伸夫` vs `Nobuo Uematsu`).
        let mut names = vec![artist_name.to_string()];
        if let Some(mb) = self.mb.as_ref() {
            if !artist_mbid.is_empty() {
                names.extend(mb.artist_names(artist_mbid, deadline).await);
            }
        }
        for name in names {
            if let Some(id) = self.piped_artist_channel(&name).await {
                if !ranked.iter().any(|(c, _)| *c == id) {
                    ranked.push((id, ChannelKey::PipedMusicArtists.as_str().to_string()));
                }
                break;
            }
        }

        let keys = ChannelKeys { ranked };

        // ⚠ **Cache only a COMPLETE answer** — and "complete" means the
        // MusicBrainz leg answered, not that the list came back non-empty.
        //
        // This is the §7.5 trap in the form that actually bit, caught in a live
        // smoke test rather than by reasoning. MusicBrainz answered `503`
        // (`musicbrainz throttled; pausing budget`), `artist_url_relations`
        // returned `None` like every other failure, the Piped key still
        // succeeded — and a **non-empty but partial** list got written to a
        // permanent cache. Daft Punk, which has both a `youtube` and a
        // `youtube music` relation, was pinned to `piped:music-artists`
        // forever: still resolving, still correct, but recorded under the
        // weakest key and unable to ever improve.
        //
        // Guarding on `!ranked.is_empty()` alone does not catch this, because
        // the degraded answer is not empty. The condition has to be about
        // whether the *authoritative* source spoke.
        if mb_answered && !keys.ranked.is_empty() {
            if let (Some(c), Ok(json)) = (self.cache.as_ref(), serde_json::to_string(&keys)) {
                let _ = c.put_yt_channel(cache_key, &json);
            }
        } else if !mb_answered {
            debug!(
                target: "meta-music",
                artist = %artist_name,
                "youtube: musicbrainz did not answer; using channel keys but NOT caching them"
            );
        }
        keys
    }

    /// Piped's `music_artists` index — the coverage-gap filler.
    ///
    /// Results are subscriber-ordered, so the first exact-name match is right
    /// in practice even where two channels share a name (two "Burial" channels
    /// exist: 37 k subscribers and 39).
    async fn piped_artist_channel(&self, artist_name: &str) -> Option<String> {
        let name = artist_name.trim();
        if name.is_empty() {
            return None;
        }
        let v = self
            .piped_get(&format!(
                "search?q={}&filter=music_artists",
                urlencoding_encode(name)
            ))
            .await?;
        let items = v.get("items")?.as_array()?;
        for it in items {
            let uploader = it
                .get("name")
                .and_then(|n| n.as_str())
                .unwrap_or_default();
            if normalise_title(uploader) == normalise_title(name) {
                if let Some(id) = it.get("url").and_then(|u| u.as_str()).and_then(channel_id_from_url) {
                    return Some(id);
                }
            }
        }
        None
    }

    /// The artist's **music videos** — the video tier of the same delegation.
    ///
    /// # Why the correctness rule is different here, and why that is sound
    ///
    /// A track is checked against a MusicBrainz *recording*: title, and a
    /// `length` to separate the album cut from the radio edit. A music video
    /// has no such anchor — MusicBrainz does not model one, so there is no
    /// canonical list to be wrong against.
    ///
    /// So the whole claim this makes is narrower, and the channel match carries
    /// it completely: **"this video was uploaded by the artist's own official
    /// channel."** That is verifiable exactly, it is the same channel key the
    /// track path already resolved, and nothing weaker is asserted — no
    /// "this is the video *of* that recording", which is the claim that would
    /// need an anchor we do not have.
    async fn music_videos(
        &self,
        query: &GatewayQuery,
        max_results: usize,
    ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
        let Some(artist_mbid) = query
            .filters
            .get("mbArtistId")
            .and_then(|v| v.first())
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
        else {
            // Id-anchored, exactly as the track path is: there is no free-text
            // browse to fall back to, and answering one would fill an artist's
            // row with somebody else's videos.
            return Ok(Vec::new());
        };
        let Some(mb) = self.mb.as_ref() else {
            return Ok(Vec::new());
        };
        let deadline = Duration::from_secs(MB_TIMEOUT_SECS);
        let Some(artist) = mb.artist(&artist_mbid, deadline).await else {
            return Ok(Vec::new());
        };

        let channels = self.channel_keys(&artist_mbid, &artist.name).await;
        if channels.ranked.is_empty() {
            debug!(
                target: "meta-music",
                artist = %artist.name,
                "youtube: no channel key; music videos would be unverifiable"
            );
            return Ok(Vec::new());
        }

        let Some(v) = self
            .piped_get(&format!(
                "search?q={}&filter=music_videos",
                urlencoding_encode(&artist.name)
            ))
            .await
        else {
            return Ok(Vec::new());
        };

        let mut out = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
        for c in parse_candidates(&v) {
            if out.len() >= max_results {
                break;
            }
            let Some((_, key)) = channels
                .ranked
                .iter()
                .find(|(id, _)| *id == c.uploader_channel)
            else {
                continue; // not this artist's channel — the entire filter
            };
            if !seen.insert(c.video_id.clone()) {
                continue;
            }
            if !self.is_embeddable(&c.video_id).await {
                continue;
            }
            if let Some(rec) =
                self.music_video_record(&artist_mbid, &artist.name, &c, key, &query.filters)
            {
                out.push(rec);
            }
        }
        Ok(out)
    }

    /// Every EXACT candidate for one track, best first.
    ///
    /// Rank is duration delta, then channel-key strength — a MusicBrainz
    /// `youtube music` relation is a stronger claim than a name match against
    /// Piped's own artist index, so it breaks ties.
    async fn exact_candidates(
        &self,
        artist: &str,
        track: &musicbrainz_feeder::client::Track,
        want_secs: Option<i64>,
        channels: &ChannelKeys,
    ) -> Vec<TrackRef> {
        let mut cands = self.search_track(artist, &track.title).await;

        // §7.5 — MusicBrainz's filed title is often not the title anyone says
        // (`Lift Yr. Skinny Fists…`, `MAP OF THE SOUL : 7`), so a single
        // exact-phrase query misses. Retry once on a normalised title before
        // concluding the corpus has nothing.
        if !cands.iter().any(|c| {
            grade(c, &track.title, want_secs, &channels.ranked).0 == Verdict::Exact
        }) {
            let loose = normalise_title(&track.title);
            if !loose.is_empty() && loose != track.title.to_lowercase() {
                cands.extend(self.search_track(artist, &loose).await);
            }
        }

        let mut out: Vec<TrackRef> = Vec::new();
        for c in cands {
            let (verdict, key) = grade(&c, &track.title, want_secs, &channels.ranked);
            if verdict != Verdict::Exact {
                continue;
            }
            let delta = want_secs.map(|w| (c.duration_secs - w).abs()).unwrap_or(i64::MAX);
            if out.iter().any(|r| r.video_id == c.video_id) {
                continue; // the two queries can return the same upload
            }
            out.push(TrackRef {
                video_id: c.video_id,
                title: c.title,
                match_key: key.unwrap_or_default(),
                delta,
            });
        }
        out.sort_by_key(|r| (r.delta, key_priority(&r.match_key)));
        out
    }

    /// Is this video embeddable **here**, right now?
    ///
    /// YouTube's oEmbed endpoint needs no key, no quota and no account:
    ///
    /// ```text
    /// 200 -> embeddable        401 -> embedding denied or private
    /// 404 -> gone
    /// ```
    ///
    /// This is what moves a dead reference from a play-time error 101/150 —
    /// the worst place to discover it — to resolve time, where it simply is not
    /// emitted. The Data API could report `regionRestriction` too, but it
    /// cannot be bought at any price (10k units/day, `search.list` = 100 units),
    /// so oEmbed carries the ubiquitous check and the Data API carries nothing.
    ///
    /// **Positives are cached, negatives are not** — same rule as the channel
    /// keys. A `401`/`404` is a real verdict, but a transport failure looks
    /// identical from here, and freezing "this video is dead" into a permanent
    /// cache on a network hiccup is the §7.5 trap wearing another hat.
    async fn is_embeddable(&self, video_id: &str) -> bool {
        let key = format!("oembed:{video_id}");
        if let Some(c) = self.cache.as_ref() {
            if matches!(c.get_yt_search(&key), Ok(Some(_))) {
                return true;
            }
        }
        let url = format!(
            "https://www.youtube.com/oembed?url={}&format=json",
            urlencoding_encode(&format!("https://www.youtube.com/watch?v={video_id}"))
        );
        let ok = match self.http.get(&url).send().await {
            Ok(r) => r.status().is_success(),
            Err(e) => {
                debug!(target: "meta-music", %video_id, error = %e, "oembed probe failed; treating as embeddable");
                // A failed probe is not a negative verdict. Emitting on doubt is
                // the right side to err on: the client already degrades a dead
                // reference gracefully (it notes and skips), whereas dropping a
                // good one is silent.
                return true;
            }
        };
        if ok {
            if let Some(c) = self.cache.as_ref() {
                let _ = c.put_yt_search(&key, "1");
            }
        }
        ok
    }

    /// Search `music_songs` for one track.
    ///
    /// `filter=music_songs` matters: it queries the YouTube *Music* corpus, so
    /// results are official audio rather than the video-first ranking of a
    /// plain search.
    async fn search_track(&self, artist: &str, title: &str) -> Vec<Candidate> {
        let q = format!("{artist} {title}");
        if let Some(c) = self.cache.as_ref() {
            if let Ok(Some(hit)) = c.get_yt_search(&meta_feeder_sdk::cache::search_key("yt", &q)) {
                if let Ok(v) = serde_json::from_str::<Value>(&hit) {
                    return parse_candidates(&v);
                }
            }
        }
        let Some(v) = self
            .piped_get(&format!(
                "search?q={}&filter=music_songs",
                urlencoding_encode(&q)
            ))
            .await
        else {
            // A failed search is NOT an empty one. Returning `[]` here would be
            // correct-looking and wrong: it would let a Piped outage present as
            // "this album has no YouTube versions". Nothing is cached, so the
            // next attempt retries.
            return Vec::new();
        };
        if let (Some(c), Ok(json)) = (self.cache.as_ref(), serde_json::to_string(&v)) {
            let _ = c.put_yt_search(&meta_feeder_sdk::cache::search_key("yt", &q), &json);
        }
        parse_candidates(&v)
    }
}

/// One recording's ranked, embeddable references, before global de-confliction.
struct TrackPick {
    recording_mbid: String,
    track: musicbrainz_feeder::client::Track,
    /// Best first: smaller duration delta wins, then the stronger channel key.
    refs: Vec<TrackRef>,
}

/// One EXACT reference, carrying what ranking and the record both need.
#[derive(Clone, Debug)]
struct TrackRef {
    video_id: String,
    title: String,
    /// Which [`ChannelKey`] produced the match, as written to `ytMatchKey`.
    match_key: String,
    /// |candidate runtime − MusicBrainz length|, in seconds. The primary rank.
    delta: i64,
}

/// **A video may belong to only one recording.**
///
/// In the study Burial's *Archangel* and *Near Dark* both resolved to
/// `VVa_aCSA9Jc` at EXACT — same channel, both durations inside tolerance.
/// They are different recordings, so at least one is wrong: a collision is
/// **proof of a bad match**, not a duplicate to tidy away. The tighter duration
/// delta keeps the claim and the weaker one drops the reference entirely rather
/// than emitting both.
///
/// A pure function over the gathered picks, deliberately: the previous form
/// mutated the output vector mid-loop and hand-patched stored indices, which is
/// exactly the shape that breaks silently once a track can contribute more than
/// one reference.
fn assign_references(picks: &mut [TrackPick]) {
    // video_id -> (pick index, winning delta)
    let mut claimed: HashMap<String, (usize, i64)> = HashMap::new();
    for (i, pick) in picks.iter().enumerate() {
        for r in &pick.refs {
            match claimed.get(&r.video_id) {
                Some(&(_, best)) if best <= r.delta => {}
                _ => {
                    claimed.insert(r.video_id.clone(), (i, r.delta));
                }
            }
        }
    }
    for (i, pick) in picks.iter_mut().enumerate() {
        pick.refs
            .retain(|r| claimed.get(&r.video_id).is_some_and(|&(owner, _)| owner == i));
    }
}

/// Grade one candidate against a track. See [`Verdict`].
fn grade(
    cand: &Candidate,
    track_title: &str,
    mb_length_secs: Option<i64>,
    channels: &[(String, String)],
) -> (Verdict, Option<String>) {
    let channel_hit = channels
        .iter()
        .find(|(id, _)| *id == cand.uploader_channel)
        .map(|(_, key)| key.clone());

    let Some(key) = channel_hit else {
        return (Verdict::NoMatch, None);
    };

    // Duration is what separates the right recording from the right artist.
    // A track MusicBrainz never timed cannot reach EXACT: with no length there
    // is nothing to compare, and "the channel agreed" alone is precisely the
    // classical trap.
    let Some(want) = mb_length_secs else {
        return (Verdict::ChannelWrongCut, Some(key));
    };
    if (cand.duration_secs - want).abs() > YT_DURATION_TOLERANCE_SECS {
        return (Verdict::ChannelWrongCut, Some(key));
    }

    // Title agreement, on top of channel+duration. Duration alone is a coin
    // flip wearing a plausible runtime — two real misses from the study, both
    // inside one second: Burial "Archangel" -> "Near Dark" (+1 s), Adele
    // "Don't You Remember" -> "Set Fire to the Rain" (+0 s).
    if !titles_agree(&cand.title, track_title) {
        return (Verdict::ChannelWrongCut, Some(key));
    }

    (Verdict::Exact, Some(key))
}

impl YouTubePlugin {
    /// One `contentKind:musicVideo` record.
    ///
    /// `fileType:video`, not `audio` — this is the video tier, and meta-listen's
    /// artist row filters on exactly that pair.
    fn music_video_record(
        &self,
        artist_mbid: &str,
        artist: &str,
        c: &Candidate,
        match_key: &str,
        query_filters: &BTreeMap<String, Vec<String>>,
    ) -> Option<DiscoveryRecord> {
        let yt_cid = meta_feeder_sdk::hash::compute_yt_video_cid("video", &c.video_id)?;

        let mut fields: BTreeMap<String, String> = BTreeMap::new();
        fields.insert("fileType".to_string(), "video".to_string());
        fields.insert("contentKind".to_string(), "musicVideo".to_string());
        // Video bytes, music domain — the case the two axes exist for.
        fields.insert("domain".to_string(), "music".to_string());
        fields.insert("workForm".to_string(), "standalone".to_string());
        fields.insert("title".to_string(), c.title.clone());
        fields.insert("artist".to_string(), artist.to_string());
        fields.insert(format!("artists/{}", artist.trim()), "true".to_string());
        fields.insert("mbArtistId".to_string(), artist_mbid.to_string());
        if c.duration_secs > 0 {
            fields.insert("fileinfo/duration".to_string(), c.duration_secs.to_string());
        }
        fields.insert(format!("cids/{yt_cid}"), "true".to_string());
        fields.insert("ytMatchKey".to_string(), match_key.to_string());
        fields.insert("youtubeid".to_string(), c.video_id.clone());
        fields.insert("source/gateway:youtube".to_string(), "true".to_string());

        // `gate::keep` requires cover + artist for a musicVideo, and a missing
        // cover drops the record silently — the same trap as the track path.
        if let Some(locator) = meta_feeder_sdk::hash::artwork_locator(&format!(
            "{YT_THUMB_BASE}/{}/hqdefault.jpg",
            c.video_id
        )) {
            fields.insert("cover".to_string(), locator);
        }

        echo_filters(&mut fields, query_filters);
        Some(DiscoveryRecord {
            upstream_id: "youtube".to_string(),
            record_id: track_record_id(artist_mbid, &c.video_id),
            fields,
        })
    }

    /// The raw tier's resolver: a free-text YouTube search, unverified.
    ///
    /// Deliberately thin. There is no channel to compare against and no
    /// duration to check, so there is nothing to *decide* — the ranking is
    /// YouTube's own, the only judgement applied is a ceiling. Anything more
    /// (a fuzzy title score, a "looks official" heuristic) would manufacture
    /// confidence this tier has no basis for and cannot label.
    async fn video_search(
        &self,
        query: &GatewayQuery,
        max_results: usize,
    ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
        // No term, no answer. A `tmdbid` alone cannot drive this path: turning
        // an id into a searchable title needs TMDB, which this feeder has no
        // client for — and inventing a term from the id would search for the
        // literal number. The anchored form is a different plugin's job (see
        // the module TODO); this one refuses rather than guessing.
        let term = query.free_text.trim().to_string();
        if term.is_empty() {
            debug!(
                target: "meta-music",
                "youtube: video query carries no free text; nothing to search on"
            );
            return Ok(Vec::new());
        }
        let cap = max_results.min(YT_VIDEO_SEARCH_MAX);
        if cap == 0 {
            return Ok(Vec::new());
        }

        let mut out = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
        for c in self.search_videos(&term, cap).await {
            if out.len() >= cap {
                break;
            }
            // Piped can repeat a video across result shapes; a duplicate record
            // id would collapse downstream anyway, but emitting it wastes a row
            // of the ceiling.
            if !seen.insert(c.video_id.clone()) {
                continue;
            }
            if let Some(r) = self.video_record(&c, &query.filters) {
                out.push(r);
            }
        }
        debug!(
            target: "meta-music",
            term = %term,
            hits = out.len(),
            "youtube: raw video search"
        );
        Ok(out)
    }

    /// `filter=videos` — the general corpus, not the Music one.
    ///
    /// The mirror of [`Self::search_track`]'s `filter=music_songs`: that filter
    /// exists to bias towards official audio, which is the right corpus for a
    /// recording and the wrong one for "a video of this thing".
    ///
    /// Cached under its **own** namespace. Sharing `search_key("yt", …)` with
    /// the track path would let a `music_songs` response answer a `videos`
    /// query whenever the two terms happened to coincide — a silent corpus swap
    /// that would look like a ranking quirk.
    async fn search_videos(&self, term: &str, take: usize) -> Vec<Candidate> {
        let key = meta_feeder_sdk::cache::search_key("yt-video", term);
        if let Some(c) = self.cache.as_ref() {
            if let Ok(Some(hit)) = c.get_yt_search(&key) {
                if let Ok(v) = serde_json::from_str::<Value>(&hit) {
                    return parse_candidates_n(&v, take);
                }
            }
        }
        let Some(v) = self
            .piped_get(&format!(
                "search?q={}&filter=videos",
                urlencoding_encode(term)
            ))
            .await
        else {
            // Same rule as the track path: a failed search is NOT an empty one,
            // and nothing is cached, so the next attempt retries. Caching a
            // Piped outage as "no such video" is the partial-answer trap that
            // pinned an artist to the wrong channel key for good.
            return Vec::new();
        };
        if let (Some(c), Ok(json)) = (self.cache.as_ref(), serde_json::to_string(&v)) {
            let _ = c.put_yt_search(&key, &json);
        }
        parse_candidates_n(&v, take)
    }

    /// One raw, unverified video record.
    ///
    /// Returns `None` if the `yt-video` cid cannot be minted — a record with no
    /// `cids/*` member is unplayable and would only add a dead row.
    fn video_record(
        &self,
        c: &Candidate,
        query_filters: &BTreeMap<String, Vec<String>>,
    ) -> Option<DiscoveryRecord> {
        let yt_cid = meta_feeder_sdk::hash::compute_yt_video_cid("video", &c.video_id)?;

        let mut fields: BTreeMap<String, String> = BTreeMap::new();
        fields.insert("fileType".to_string(), "video".to_string());
        // **No `contentKind`.** The music paths know what they resolved; this
        // one genuinely does not, and a guess would be worse than a gap:
        // `contentKind:pack` in particular is dropped outright by meta-watch's
        // raw projection, and any wrong value narrows the record out of filters
        // it should have matched. An absent axis is accepted everywhere.
        fields.insert("title".to_string(), c.title.clone());
        fields.insert(format!("cids/{yt_cid}"), "true".to_string());
        fields.insert("youtubeid".to_string(), c.video_id.clone());
        fields.insert("source/gateway:youtube".to_string(), "true".to_string());
        if c.duration_secs > 0 {
            fields.insert("fileinfo/duration".to_string(), c.duration_secs.to_string());
        }
        // The confidence signal, stated rather than implied. Every other record
        // this feeder emits carries the channel key that verified it; this one
        // carries the fact that nothing did. A consumer that later wants to
        // rank or hide the raw tier has the field to do it with, and no reader
        // has to infer "unverified" from the *absence* of a key.
        fields.insert("ytMatchKey".to_string(), RAW_MATCH_KEY.to_string());
        // Artwork + synopsis. Neither is needed for the raw bucket — meta-watch
        // renders those rows without a poster — but a record outlives the query
        // that produced it, and a consumer applying a poster+description gate
        // (meta-listen's, or meta-watch's enriched grid should this hit ever be
        // anchored by something else) would otherwise drop it silently. That
        // silence is the expensive failure mode here, so pay two fields now.
        //
        // `poster_url`, not `cover`: `cover` is the music tier's sleeve
        // vocabulary and this is a video record. The core fetches and seeds a
        // `*_url` field (`remote_feeder::SEEDABLE_FIELDS`), which is why the
        // per-query ceiling exists.
        fields.insert(
            "poster_url".to_string(),
            format!("{YT_THUMB_BASE}/{}/hqdefault.jpg", c.video_id),
        );
        let uploader = c.uploader.trim();
        fields.insert(
            "description/eng".to_string(),
            if uploader.is_empty() {
                "Delegated playback — YouTube. Unverified search result.".to_string()
            } else {
                format!("Delegated playback — YouTube, {uploader}. Unverified search result.")
            },
        );
        if !uploader.is_empty() {
            fields.insert("uploader".to_string(), uploader.to_string());
        }

        // Echoes the query's own filters onto the record — which is how a
        // `tmdbid:`-carrying query gets that id stamped on the hit, tying it to
        // the work the consumer asked about. Note what this is NOT: echoing an
        // id is not *verifying* one, and it deliberately does not set
        // `anchored`. A record that claimed the anchor it merely inherited is
        // the leak that once made a whole indexer's results look trusted.
        echo_filters(&mut fields, query_filters);
        Some(DiscoveryRecord {
            upstream_id: "youtube".to_string(),
            record_id: video_record_id(&c.video_id),
            fields,
        })
    }
}

/// Rank order for [`ChannelKey`] as a tie-break: authoritative first.
fn key_priority(match_key: &str) -> u8 {
    match match_key {
        "mb:youtube-music" => 0,
        "mb:youtube" => 1,
        _ => 2,
    }
}

/// Do two titles name the same recording?
///
/// Two rules, and the asymmetry between the costs is what shapes them: a false
/// *refusal* costs one row of coverage, a false *accept* puts the wrong song
/// under the right name. So this stays deliberately tight.
///
/// 1. **Containment, either direction** — because YouTube Music decorates a
///    title with the feature credits MusicBrainz keeps in the artist credit
///    ("Get Lucky" vs "Get Lucky (feat. Pharrell Williams and Nile Rodgers)").
/// 2. **High token overlap on longer titles** — because MusicBrainz's filed
///    title is often not the title anyone says, and the difference is a single
///    word: `Lift Yr. Skinny Fists…` against `Lift Your Skinny Fists…` shares
///    nine tokens of ten but fails containment outright.
///
/// Rule 2 is gated at [`TITLE_MIN_TOKENS`] tokens precisely so it cannot loosen
/// short titles, where a single shared word is most of the title and overlap
/// stops discriminating.
fn titles_agree(a: &str, b: &str) -> bool {
    let (na, nb) = (normalise_title(a), normalise_title(b));
    if na.is_empty() || nb.is_empty() {
        return false;
    }
    if na.contains(&nb) || nb.contains(&na) {
        return true;
    }
    let ta: std::collections::BTreeSet<&str> = na.split(' ').collect();
    let tb: std::collections::BTreeSet<&str> = nb.split(' ').collect();
    let smaller = ta.len().min(tb.len());
    if smaller < TITLE_MIN_TOKENS {
        return false;
    }
    let shared = ta.intersection(&tb).count();
    shared as f32 / smaller as f32 >= TITLE_MIN_OVERLAP
}

/// Token count below which [`titles_agree`] rule 2 does not apply.
const TITLE_MIN_TOKENS: usize = 4;
/// Fraction of the shorter title's tokens that must be shared under rule 2.
const TITLE_MIN_OVERLAP: f32 = 0.8;

/// Lowercase, drop parentheticals, strip non-alphanumerics, collapse spaces.
///
/// MusicBrainz's title is often not the title anyone says — `Lift Yr. Skinny
/// Fists Like Antennas to Heaven!` and `MAP OF THE SOUL : 7` are the filed
/// forms — so an exact-phrase comparison misses both.
fn normalise_title(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    let mut depth = 0usize;
    for ch in s.chars() {
        match ch {
            '(' | '[' => depth += 1,
            ')' | ']' => depth = depth.saturating_sub(1),
            _ if depth > 0 => {}
            c if c.is_alphanumeric() => out.extend(c.to_lowercase()),
            _ => out.push(' '),
        }
    }
    out.split_whitespace().collect::<Vec<_>>().join(" ")
}

/// Extract a bare channel id from any of the URL shapes MusicBrainz and Piped
/// use (`https://www.youtube.com/channel/UC…`, `/channel/UC…`).
fn channel_id_from_url(url: &str) -> Option<String> {
    let (_, tail) = url.split_once("/channel/")?;
    let id = tail.split(['/', '?', '#']).next()?.trim();
    (!id.is_empty()).then(|| id.to_string())
}

/// Extract a bare video id from Piped's `/watch?v=<id>` url field.
fn video_id_from_url(url: &str) -> Option<String> {
    let (_, tail) = url.split_once("v=")?;
    let id = tail.split(['&', '#']).next()?.trim();
    (!id.is_empty()).then(|| id.to_string())
}

fn parse_candidates(v: &Value) -> Vec<Candidate> {
    parse_candidates_n(v, YT_CANDIDATES_PER_TRACK)
}

/// [`parse_candidates`] with an explicit ceiling.
///
/// The music paths want [`YT_CANDIDATES_PER_TRACK`] — a *verification* budget,
/// sized by how deep into the ranking a correct match plausibly sits. The raw
/// tier's ceiling is a different quantity entirely (how many rows to show), so
/// it passes its own rather than inheriting a constant that means something
/// else. Same parse either way: `filter=videos` and `filter=music_*` return the
/// same item shape.
fn parse_candidates_n(v: &Value, take: usize) -> Vec<Candidate> {
    let Some(items) = v.get("items").and_then(|i| i.as_array()) else {
        return Vec::new();
    };
    items
        .iter()
        .take(take)
        .filter_map(|it| {
            Some(Candidate {
                video_id: video_id_from_url(it.get("url")?.as_str()?)?,
                title: it.get("title")?.as_str()?.to_string(),
                uploader_channel: it
                    .get("uploaderUrl")
                    .and_then(|u| u.as_str())
                    .and_then(channel_id_from_url)
                    .unwrap_or_default(),
                duration_secs: it.get("duration").and_then(|d| d.as_i64()).unwrap_or(-1),
                uploader: it
                    .get("uploaderName")
                    .and_then(|u| u.as_str())
                    .unwrap_or_default()
                    .to_string(),
            })
        })
        .collect()
}

/// Minimal percent-encoder for a query-string value.
fn urlencoding_encode(s: &str) -> String {
    let mut out = String::with_capacity(s.len() * 3);
    for b in s.as_bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(*b as char)
            }
            b' ' => out.push('+'),
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

/// `<mbRecordingId>~<videoId>` — the record id this plugin mints.
///
/// # Why the video id is *in* the id
///
/// It makes [`YouTubePlugin::compute_outcomes`] a **pure function with zero
/// I/O**, exactly as `compute_card_cid` is for the card feeder. The alternative
/// — keying only on the recording and re-resolving at outcome time — would
/// re-run a MusicBrainz tracklist, a channel lookup and a Piped search, and
/// could legitimately pick a *different* video than the one the consumer is
/// looking at. Minting a reference other than the one on screen is worse than
/// slow.
///
/// It also gives the multi-reference design its natural shape: several ranked
/// references for one recording are several records, each stably addressed,
/// each converging across peers via the `cids/` reverse index — rather than one
/// record whose meaning silently changes when the resolver's pick moves.
pub fn track_record_id(recording_mbid: &str, video_id: &str) -> String {
    format!("{recording_mbid}~{video_id}")
}

/// Inverse of [`track_record_id`] — `(recording_mbid, video_id)`.
pub fn split_track_record_id(record_id: &str) -> Option<(&str, &str)> {
    let (rec, vid) = record_id.split_once('~')?;
    (!rec.is_empty() && !vid.is_empty()).then_some((rec, vid))
}

/// `yt~<videoId>` — the record id the **raw** tier mints.
///
/// Shares [`split_track_record_id`]'s shape on purpose, so `compute_outcomes`
/// stays one zero-I/O function serving both tiers: it only ever needs the right
/// half. The left half is [`VIDEO_RECORD_NS`] rather than a second MBID because
/// a raw hit has no recording behind it — and a constant makes the id a pure
/// function of the video, so the same video found via two different search
/// terms dedupes to one record instead of two.
pub fn video_record_id(video_id: &str) -> String {
    format!("{VIDEO_RECORD_NS}~{video_id}")
}

/// Did the query **explicitly** ask for this file type?
///
/// Explicit, unlike [`wants_content_kind`]: an absent axis means "no
/// preference", and treating that as a video request would let the raw tier
/// answer questions nobody pointed at it. The music paths can afford the
/// permissive reading because an id anchor follows; this one cannot.
fn requests_file_type(query: &GatewayQuery, file_type: &str) -> bool {
    query
        .filters
        .get("fileType")
        .is_some_and(|v| v.iter().any(|x| x.trim().eq_ignore_ascii_case(file_type)))
}

/// Does the query carry a MusicBrainz anchor of any kind?
///
/// The guard that keeps the raw tier out of the verified tiers' way. If a
/// caller handed us an id, there is a path that can *verify* an answer, and
/// serving a free-text guess instead would be a silent downgrade — the kind
/// that shows up as "it used to be right".
fn has_musicbrainz_anchor(query: &GatewayQuery) -> bool {
    ["mbReleaseGroupId", "mbArtistId", "mbRecordingId"]
        .iter()
        .any(|k| {
            query
                .filters
                .get(*k)
                .is_some_and(|v| v.iter().any(|s| !s.trim().is_empty()))
        })
}

#[async_trait]
impl FeederPlugin for YouTubePlugin {
    fn upstream_id(&self) -> &'static str {
        "youtube"
    }

    fn served_file_types(&self) -> &'static [&'static str] {
        &["audio", "video"]
    }

    /// `track` and `musicVideo` — deliberately **not** `pack`.
    ///
    /// A pack is a container whose members are separately addressable, and
    /// there is no such thing as a delegated *album*: each reference is one
    /// video. Claiming `pack` would advertise a capability that can only
    /// answer with a synthetic container holding nothing fetchable.
    ///
    /// `musicVideo` is the *video* tier of the same delegation, and it is what
    /// finally fills meta-listen's artist-page "Music videos" row — empty until
    /// now because nothing in the platform produced the content kind. YouTube
    /// hosts the Art Track and the music video as **separate video ids**, which
    /// is exactly what makes the YouTube-Music-style song/video toggle
    /// implementable: swap the reference, keep the position.
    fn served_content_kinds(&self) -> &'static [&'static str] {
        &["track", "musicVideo"]
    }

    fn configure(&mut self, cache_dir: &Path) -> Result<(), ConfigError> {
        let cache = MidhashCache::open(cache_dir).map_err(|e| ConfigError::Other {
            plugin: "youtube",
            source: anyhow::anyhow!("open music cache: {e}"),
        })?;
        // Its own MusicBrainz budget rather than a shared one: MusicBrainz's
        // 1 req/s ceiling is enforced per source IP, and a budget shared with
        // the card plugin would let a board refresh and an album open burst
        // past it together.
        let budget = RateBudget::new(MB_RATE_PER_SEC, MB_BURST);
        let mb = match &self.mb_base {
            Some(base) => MusicBrainzClient::with_base(base.clone(), budget, cache.clone()),
            None => MusicBrainzClient::new(budget, cache.clone()),
        };
        self.mb = Some(Arc::new(mb));
        self.cache = Some(cache);
        Ok(())
    }

    fn health(&self) -> PluginHealth {
        match self.mb {
            Some(_) => PluginHealth::Ok,
            None => PluginHealth::Degraded {
                reason: "configure() not yet called".to_string(),
            },
        }
    }

    /// Resolve one release-group's tracks to YouTube references.
    ///
    /// # This plugin is the mirror image of `internetarchive`
    ///
    /// The archive holds no MusicBrainz ids, so it must return **nothing** for
    /// an id-anchored query. This plugin can *only* work from one: you cannot
    /// search YouTube for a UUID, so the chain is necessarily
    /// `mbReleaseGroupId` → tracklist (titles + `length` + `mbRecordingId`) →
    /// artist channel → per-track search → filter. A free-text query therefore
    /// returns nothing here — not because it would be expensive, but because
    /// the resolver has no anchor to verify a match against, and an unverified
    /// match is the one output this tier must never produce.
    async fn handle_query(
        &self,
        query: &GatewayQuery,
        max_results: usize,
    ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
        if !meta_feeder_sdk::query_eval::query_accepts_plugin(
            query,
            self.served_file_types(),
            self.served_content_kinds(),
        ) {
            return Ok(Vec::new());
        }
        // Two anchors, two answers. The music-video row asks
        // `mbArtistId:<uuid> fileType:video contentKind:musicVideo`, which is a
        // different question from an album page's
        // `mbReleaseGroupId:<uuid> fileType:audio` — different entity, different
        // corpus, different correctness criterion. Dispatch on which anchor is
        // present rather than trying to serve both from one path.
        if wants_content_kind(query, "musicVideo") && !wants_content_kind(query, "track") {
            return self.music_videos(query, max_results).await;
        }
        // **The raw video tier.** Everything above this line is id-anchored and
        // refuses far more than it emits; this branch is the deliberate
        // exception, and the reason it is safe is *structural*, not a matter of
        // resolver discipline.
        //
        // A consumer's untyped free-text search never reaches a gateway at all
        // (meta-search's load gate), so the query that lands here has already
        // said `fileType:video` — a video client asking a video question. What
        // it cannot say is *which* work it means, and there is no anchor to
        // verify a match against. So this path emits **unverified** hits, which
        // every other path in this module exists to prevent.
        //
        // What makes that honest rather than a regression: these records carry
        // no `anchored` flag, and meta-watch's `release_from_record` drops every
        // non-anchored gateway hit from the enriched grid *unconditionally* —
        // poster and synopsis do not buy a way in. They can therefore only
        // surface in the "raw sources" section, which is precisely the UI
        // affordance for "found it, can't vouch for it". The tier is
        // second-class by construction, not by convention.
        //
        // Ordering matters here and is load-bearing:
        //
        // * `musicVideo` is matched **above**, so meta-listen's artist row
        //   (`mbArtistId:… fileType:video contentKind:musicVideo`) still takes
        //   the verified path and never falls through to this one.
        // * The `mbReleaseGroupId` track path below is `fileType:audio`, so it
        //   cannot reach this branch either.
        // * `has_musicbrainz_anchor` is belt-and-braces for both: an id-anchored
        //   query has something better than a free-text guess available, and
        //   answering it from here would quietly downgrade a verified tier.
        if requests_file_type(query, "video") && !has_musicbrainz_anchor(query) {
            return self.video_search(query, max_results).await;
        }
        if !wants_content_kind(query, "track") {
            return Ok(Vec::new());
        }

        let Some(rg_mbid) = query
            .filters
            .get("mbReleaseGroupId")
            .and_then(|v| v.first())
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
        else {
            debug!(
                target: "meta-music",
                "youtube: query carries no mbReleaseGroupId; nothing to anchor on"
            );
            return Ok(Vec::new());
        };

        let Some(mb) = self.mb.as_ref() else {
            return Ok(Vec::new());
        };
        let deadline = Duration::from_secs(MB_TIMEOUT_SECS);

        let Some(release) = mb.tracklist(&rg_mbid, deadline).await else {
            // Not "this album has no tracks" — the lookup did not answer. Never
            // conflate the two (study §7.5); emit nothing and let the next
            // attempt retry.
            debug!(target: "meta-music", rg = %rg_mbid, "youtube: no tracklist");
            return Ok(Vec::new());
        };

        // Whose channel to look for. The release-group's credited artist is the
        // right *starting* point, but see `Verdict::ChannelWrongCut`: for
        // classical this is the composer, and the duration filter is what stops
        // that from resolving to someone else's performance.
        let rg = mb.release_group(&rg_mbid, deadline).await;
        let (artist_mbid, artist_name) = rg
            .as_ref()
            .and_then(|g| primary_artist(&g.artist_credit))
            .map(|a| (a.id.clone(), a.name.clone()))
            .unwrap_or_default();

        let channels = self.channel_keys(&artist_mbid, &artist_name).await;
        if channels.ranked.is_empty() {
            debug!(
                target: "meta-music",
                artist = %artist_name,
                "youtube: no channel key for this artist; every track would be unverifiable"
            );
            return Ok(Vec::new());
        }

        let album_title = release.title.clone();

        // ── 1. Gather ranked EXACT candidates per track ──────────────────────
        //
        // Ranked, not reduced to a single best: §3's rot mitigation is to carry
        // *several* references per recording so a client can fall through when
        // one is taken down, geo-blocked or has embedding disabled. A single
        // reference means a takedown kills the track outright.
        let mut picks: Vec<TrackPick> = Vec::new();
        for medium in &release.media {
            for track in &medium.tracks {
                let recording_mbid = track
                    .recording
                    .as_ref()
                    .map(|r| r.id.clone())
                    .unwrap_or_default();
                if recording_mbid.is_empty() {
                    // Without it the consumer matches versions to track rows by
                    // fuzzy title and lands on the wrong row for anything with a
                    // reprise, a remix or a live variant.
                    continue;
                }
                let want_secs = track
                    .length
                    .or_else(|| track.recording.as_ref().and_then(|r| r.length))
                    .map(|ms| (ms / 1000) as i64);

                let mut refs = self.exact_candidates(&artist_name, track, want_secs, &channels).await;
                // Probe only what could plausibly be emitted — one spare, so a
                // single dead reference does not cost the track a slot.
                refs.truncate(MAX_REFS_PER_RECORDING + 1);
                let mut embeddable = Vec::new();
                for r in refs {
                    if self.is_embeddable(&r.video_id).await {
                        embeddable.push(r);
                    }
                }
                if embeddable.is_empty() {
                    continue;
                }
                picks.push(TrackPick {
                    recording_mbid,
                    track: track.clone(),
                    refs: embeddable,
                });
            }
        }

        // ── 2. One video may belong to only one recording ────────────────────
        assign_references(&mut picks);

        // ── 3. Emit ─────────────────────────────────────────────────────────
        let mut out: Vec<DiscoveryRecord> = Vec::new();
        for pick in &picks {
            for (rank, r) in pick.refs.iter().take(MAX_REFS_PER_RECORDING).enumerate() {
                if out.len() >= max_results {
                    break;
                }
                if let Some(rec) = self.track_record(
                    &rg_mbid,
                    &pick.recording_mbid,
                    &album_title,
                    &artist_name,
                    &pick.track,
                    r,
                    rank as u32,
                    &query.filters,
                ) {
                    out.push(rec);
                }
            }
        }

        out.truncate(max_results);
        Ok(out)
    }

    /// Metadata-only: a reference, never bytes.
    ///
    /// The `(None, Some)` branch — `bytes: None`, so the core auto-stores the
    /// record and stops. There is no blockstore seed (gated on `Sha2_256`), no
    /// WebDAV blob, and no `file_extension` (which MUST be `None` when `bytes`
    /// is).
    /// Serves **both** tiers: the left half of the id is whatever minted it (a
    /// MusicBrainz recording, or [`VIDEO_RECORD_NS`] for a raw hit) and is
    /// deliberately unread — the cid is a pure function of the video id alone,
    /// so there is nothing here to branch on.
    async fn compute_outcomes(&self, record_id: &str) -> Result<Vec<HashOutcome>, GatewayError> {
        let (_minted_by, video_id) = split_track_record_id(record_id).ok_or_else(|| {
            GatewayError::Permanent(format!("malformed youtube record_id '{record_id}'"))
        })?;
        let hash = meta_feeder_sdk::hash::compute_yt_video_cid("video", video_id).ok_or_else(
            || GatewayError::Permanent(format!("video id too long to encode: '{video_id}'")),
        )?;

        // `record: None` — unlike the card feeder there is nothing to
        // re-resolve. `handle_query` already wrote the full record, and the
        // only thing this cid could be re-derived from is the video id already
        // in the record_id.
        Ok(vec![HashOutcome {
            hash: Hash(hash),
            hash_kind: HashKind::YtVideo,
            bytes: None,
            record: None,
            file_extension: None,
        }])
    }

    /// Never. This tier has no byte path by construction.
    async fn handle_fetch(&self, _record_id: &str) -> Result<Option<ByteStream>, GatewayError> {
        Ok(None)
    }
}

impl YouTubePlugin {
    /// Build one delegated-playback track record.
    ///
    /// Returns `None` if the `yt-video` cid cannot be minted — a record with no
    /// `cids/*` member is unplayable and would only add a dead row.
    #[allow(clippy::too_many_arguments)]
    fn track_record(
        &self,
        rg_mbid: &str,
        recording_mbid: &str,
        album: &str,
        artist: &str,
        track: &musicbrainz_feeder::client::Track,
        r: &TrackRef,
        rank: u32,
        query_filters: &BTreeMap<String, Vec<String>>,
    ) -> Option<DiscoveryRecord> {
        let yt_cid = meta_feeder_sdk::hash::compute_yt_video_cid("video", &r.video_id)?;

        let mut fields: BTreeMap<String, String> = BTreeMap::new();
        fields.insert("fileType".to_string(), "audio".to_string());
        fields.insert("contentKind".to_string(), "track".to_string());
        fields.insert("domain".to_string(), "music".to_string());
        fields.insert("workForm".to_string(), "standalone".to_string());
        fields.insert("title".to_string(), track.title.clone());
        fields.insert("album".to_string(), album.to_string());
        if !artist.trim().is_empty() {
            fields.insert("albumArtist".to_string(), artist.to_string());
            fields.insert("artist".to_string(), artist.to_string());
            fields.insert(format!("artists/{}", artist.trim()), "true".to_string());
        }
        if track.position > 0 {
            fields.insert("trackNumber".to_string(), track.position.to_string());
        }
        // Matched on by the consumer to place a version against a track row.
        fields.insert("mbRecordingId".to_string(), recording_mbid.to_string());
        fields.insert("mbReleaseGroupId".to_string(), rg_mbid.to_string());

        // The reference itself.
        fields.insert(format!("cids/{yt_cid}"), "true".to_string());
        // Which key produced the channel match. See `ChannelKey`.
        fields.insert("ytMatchKey".to_string(), r.match_key.clone());
        // Which fallback position this reference occupies for the recording,
        // 0 = best. Several references per recording is the mitigation for
        // reference rot (§3): the id is immutable but its *availability* is
        // not, so the client needs a defined order to fall through.
        fields.insert("ytRank".to_string(), rank.to_string());

        // The canonical `<upstream_id>id` field every feeder record must carry.
        fields.insert("youtubeid".to_string(), r.video_id.clone());
        fields.insert("source/gateway:youtube".to_string(), "true".to_string());

        // The cover — load-bearing, and silently fatal if omitted.
        //
        // The consumer's gate requires artwork (`gate::has_cover`), so a record
        // without one is dropped before it ever reaches a version list, with no
        // error anywhere. Emitted as a `url` LOCATOR (`0x1006`), which is the
        // current convention: `cover_url` was REMOVED on 2026-08-21 because a
        // `*_url` field only looks transient — the gateway rewrites it on
        // success, so a 404 leaves the raw URL persisted forever.
        //
        // Note this is `cover`, not `poster`: music artwork has its own
        // vocabulary since the two images come from different places
        // (METADATA_KEYS.md §6).
        if let Some(locator) = meta_feeder_sdk::hash::artwork_locator(&format!(
            "{YT_THUMB_BASE}/{}/hqdefault.jpg",
            r.video_id
        )) {
            fields.insert("cover".to_string(), locator);
        }

        echo_filters(&mut fields, query_filters);
        Some(DiscoveryRecord {
            upstream_id: "youtube".to_string(),
            record_id: track_record_id(recording_mbid, &r.video_id),
            fields,
        })
    }
}

/// Local copies of the two tiny query helpers `internetarchive` also uses.
/// Duplicated rather than shared because they are three lines each and the
/// alternative is a `pub` surface on a sibling plugin module.
fn wants_content_kind(query: &GatewayQuery, kind: &str) -> bool {
    match query.filters.get("contentKind") {
        None => true,
        Some(values) if values.is_empty() => true,
        Some(values) => values.iter().any(|v| v.trim().eq_ignore_ascii_case(kind)),
    }
}

fn echo_filters(fields: &mut BTreeMap<String, String>, query_filters: &BTreeMap<String, Vec<String>>) {
    for (key, allowed) in query_filters {
        if key == "languages" || key == "genres" || fields.contains_key(key) || allowed.is_empty() {
            continue;
        }
        fields.insert(key.clone(), allowed.join(","));
    }
}

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

    fn chan(id: &str) -> Vec<(String, String)> {
        vec![(id.to_string(), ChannelKey::MbYoutubeMusic.as_str().to_string())]
    }

    fn cand(video: &str, title: &str, channel: &str, secs: i64) -> Candidate {
        Candidate {
            video_id: video.to_string(),
            title: title.to_string(),
            uploader_channel: channel.to_string(),
            duration_secs: secs,
            uploader: String::new(),
        }
    }

    #[test]
    fn exact_needs_channel_and_duration_and_title() {
        // The real Daft Punk case from the study: MB "Get Lucky" at 370 s, the
        // YouTube Music upload decorated with the feature credits.
        let c = cand(
            "4D7u5KF7SP8",
            "Get Lucky (feat. Pharrell Williams and Nile Rodgers)",
            "UCRr1xG_2WIDs18a6cIiCxeA",
            370,
        );
        let (v, key) = grade(&c, "Get Lucky", Some(370), &chan("UCRr1xG_2WIDs18a6cIiCxeA"));
        assert_eq!(v, Verdict::Exact);
        assert_eq!(key.as_deref(), Some("mb:youtube-music"));
    }

    #[test]
    fn a_different_channel_is_never_a_match() {
        let c = cand("x", "Get Lucky", "UCsomeoneelse", 370);
        let (v, _) = grade(&c, "Get Lucky", Some(370), &chan("UCRr1xG_2WIDs18a6cIiCxeA"));
        assert_eq!(v, Verdict::NoMatch);
    }

    #[test]
    fn channel_match_with_far_off_duration_is_a_refusal() {
        // THE classical trap, with the study's measured deltas. MusicBrainz
        // credits Bach for a Glenn Gould recording, so the channel agrees
        // confidently and returns someone else's performance (+32 s, +209 s).
        for delta in [32, 209] {
            let c = cand("v", "The Goldberg Variations", "UCbach", 1000 + delta);
            let (v, key) = grade(&c, "The Goldberg Variations", Some(1000), &chan("UCbach"));
            assert_eq!(
                v,
                Verdict::ChannelWrongCut,
                "a +{delta}s channel match must refuse, never widen into a near-miss"
            );
            // The key is still reported so the refusal is attributable.
            assert!(key.is_some());
        }
    }

    #[test]
    fn a_track_musicbrainz_never_timed_cannot_reach_exact() {
        // No length => nothing to compare => "the channel agreed" alone, which
        // is exactly the shape the classical trap produces.
        let c = cand("v", "Some Track", "UCartist", 200);
        let (v, _) = grade(&c, "Some Track", None, &chan("UCartist"));
        assert_eq!(v, Verdict::ChannelWrongCut);
    }

    #[test]
    fn duration_alone_is_not_enough_the_title_must_agree() {
        // Both real misses from the study's first run, both inside one second.
        let burial = cand("v1", "Near Dark", "UCburial", 241);
        let (v, _) = grade(&burial, "Archangel", Some(240), &chan("UCburial"));
        assert_eq!(v, Verdict::ChannelWrongCut);

        let adele = cand("v2", "Set Fire to the Rain", "UCadele", 242);
        let (v, _) = grade(&adele, "Don't You Remember", Some(242), &chan("UCadele"));
        assert_eq!(v, Verdict::ChannelWrongCut);
    }

    #[test]
    fn title_normalisation_survives_musicbrainz_filing_conventions() {
        // MusicBrainz's title is often not the title anyone says.
        assert!(titles_agree(
            "Lift Your Skinny Fists Like Antennas to Heaven",
            "Lift Yr. Skinny Fists Like Antennas to Heaven!"
        ));
        assert!(titles_agree("MAP OF THE SOUL : 7", "Map of the Soul: 7"));
        // ...but it is not so loose that two different songs agree.
        assert!(!titles_agree("Archangel", "Near Dark"));
        assert!(!titles_agree("Don't You Remember", "Set Fire to the Rain"));
        // The short-title guard: overlap must NOT rescue a 3-token near-miss,
        // where one shared word is most of the title.
        assert!(!titles_agree("Love Me Tender", "Love Me Do"));
    }

    #[test]
    fn channel_and_video_ids_are_extracted_from_every_url_shape() {
        assert_eq!(
            channel_id_from_url("https://www.youtube.com/channel/UC_kRDKYrUlrbtrSiyu5Tflg").as_deref(),
            Some("UC_kRDKYrUlrbtrSiyu5Tflg")
        );
        assert_eq!(
            channel_id_from_url("https://music.youtube.com/channel/UCRr1xG_2WIDs18a6cIiCxeA")
                .as_deref(),
            Some("UCRr1xG_2WIDs18a6cIiCxeA")
        );
        assert_eq!(channel_id_from_url("/channel/UCabc?x=1").as_deref(), Some("UCabc"));
        assert!(channel_id_from_url("https://youtube.com/user/foo").is_none());

        assert_eq!(video_id_from_url("/watch?v=4D7u5KF7SP8").as_deref(), Some("4D7u5KF7SP8"));
        assert_eq!(
            video_id_from_url("/watch?v=4D7u5KF7SP8&list=OLAK").as_deref(),
            Some("4D7u5KF7SP8")
        );
    }

    #[test]
    fn record_id_round_trips_and_mints_the_cid_with_no_io() {
        let id = track_record_id("345e4a72-46b4-48ba-8541-27c6b23c3b8c", "4D7u5KF7SP8");
        let (rec, vid) = split_track_record_id(&id).expect("round-trips");
        assert_eq!(rec, "345e4a72-46b4-48ba-8541-27c6b23c3b8c");
        assert_eq!(vid, "4D7u5KF7SP8");
        // The same cid the golden vector pins.
        assert_eq!(
            meta_feeder_sdk::hash::compute_yt_video_cid("video", vid).unwrap(),
            "bagecaaarav3gszdfn42ein3vgvfumn2tka4a"
        );
        assert!(split_track_record_id("no-separator").is_none());
        assert!(split_track_record_id("rec~").is_none());
    }

    #[tokio::test]
    async fn compute_outcomes_is_metadata_only() {
        let p = YouTubePlugin::default();
        let id = track_record_id("rec-mbid", "4D7u5KF7SP8");
        let out = p.compute_outcomes(&id).await.expect("outcome");
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].hash_kind, HashKind::YtVideo);
        // The three properties that make this tier what it is.
        assert!(out[0].bytes.is_none(), "delegated playback transports no bytes");
        assert!(
            out[0].file_extension.is_none(),
            "file_extension MUST be None when bytes is None"
        );
        assert_eq!(out[0].hash.0, "bagecaaarav3gszdfn42ein3vgvfumn2tka4a");
    }

    #[tokio::test]
    async fn handle_fetch_never_returns_a_stream() {
        let p = YouTubePlugin::default();
        assert!(p
            .handle_fetch(&track_record_id("rec", "vid"))
            .await
            .expect("ok")
            .is_none());
    }

    #[tokio::test]
    async fn a_query_with_no_release_group_anchor_returns_nothing() {
        // The mirror image of `internetarchive`: this plugin can ONLY work from
        // an id anchor, because you cannot search YouTube for a UUID and an
        // unverified match is the one output this tier must never produce.
        let p = YouTubePlugin::default();
        let q = GatewayQuery {
            raw_text: "daft punk".to_string(),
            free_text: "daft punk".to_string(),
            filters: BTreeMap::new(),
            ranges: Vec::new(),
            negations: Vec::new(),
        };
        assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
    }

    // --- the raw video tier -------------------------------------------------

    fn vq(free_text: &str, filters: &[(&str, &[&str])]) -> GatewayQuery {
        GatewayQuery {
            raw_text: free_text.to_string(),
            free_text: free_text.to_string(),
            filters: filters
                .iter()
                .map(|(k, vs)| {
                    (
                        (*k).to_string(),
                        vs.iter().map(|v| (*v).to_string()).collect::<Vec<_>>(),
                    )
                })
                .collect(),
            ranges: Vec::new(),
            negations: Vec::new(),
        }
    }

    #[test]
    fn file_type_request_is_explicit_not_permissive() {
        // The asymmetry with `wants_content_kind` is the point: an absent axis
        // means "no preference", and the raw tier must not read that as an
        // invitation.
        assert!(requests_file_type(
            &vq("x", &[("fileType", &["video"])]),
            "video"
        ));
        assert!(requests_file_type(
            &vq("x", &[("fileType", &["audio", "video"])]),
            "video"
        ));
        assert!(!requests_file_type(&vq("x", &[]), "video"));
        assert!(!requests_file_type(
            &vq("x", &[("fileType", &["audio"])]),
            "video"
        ));
    }

    #[test]
    fn musicbrainz_anchor_is_detected_on_every_id_axis() {
        for k in ["mbReleaseGroupId", "mbArtistId", "mbRecordingId"] {
            assert!(
                has_musicbrainz_anchor(&vq("x", &[(k, &["8bfac288-ccc5-445d-9a7e-1a1"])])),
                "{k} should count as an anchor"
            );
        }
        assert!(!has_musicbrainz_anchor(&vq("x", &[("tmdbid", &["95479"])])));
        // Present but empty is not an anchor — it cannot verify anything.
        assert!(!has_musicbrainz_anchor(&vq("x", &[("mbArtistId", &[""])])));
    }

    #[test]
    fn video_record_id_round_trips_through_the_shared_splitter() {
        // `compute_outcomes` serves both tiers off this one parse; a raw id it
        // cannot split would be a permanent error at hash time, long after the
        // record looked fine.
        let id = video_record_id("4D7u5KF7SP8");
        let (ns, vid) = split_track_record_id(&id).expect("splits");
        assert_eq!(ns, VIDEO_RECORD_NS);
        assert_eq!(vid, "4D7u5KF7SP8");
    }

    #[tokio::test]
    async fn compute_outcomes_serves_a_raw_record_id() {
        let p = YouTubePlugin::default();
        let out = p
            .compute_outcomes(&video_record_id("4D7u5KF7SP8"))
            .await
            .expect("ok");
        assert_eq!(out.len(), 1);
        assert_eq!(
            out[0].hash.0,
            meta_feeder_sdk::hash::compute_yt_video_cid("video", "4D7u5KF7SP8").unwrap()
        );
        assert!(out[0].bytes.is_none(), "the tier has no byte path");
    }

    #[test]
    fn raw_record_carries_the_fields_a_consumer_gates_on() {
        let p = YouTubePlugin::default();
        let c = Candidate {
            video_id: "4D7u5KF7SP8".to_string(),
            title: "Three-Body EP01".to_string(),
            uploader_channel: "UCabc".to_string(),
            duration_secs: 2712,
            uploader: "Tencent Video".to_string(),
        };
        let filters: BTreeMap<String, Vec<String>> = [
            ("fileType".to_string(), vec!["video".to_string()]),
            ("tmdbid".to_string(), vec!["95479".to_string()]),
        ]
        .into_iter()
        .collect();
        let r = p.video_record(&c, &filters).expect("record");
        let f = &r.fields;

        assert_eq!(f.get("fileType").map(String::as_str), Some("video"));
        assert_eq!(f.get("youtubeid").map(String::as_str), Some("4D7u5KF7SP8"));
        assert_eq!(f.get("ytMatchKey").map(String::as_str), Some(RAW_MATCH_KEY));
        assert_eq!(f.get("fileinfo/duration").map(String::as_str), Some("2712"));
        assert!(
            f.contains_key("source/gateway:youtube"),
            "raw bucket needs it"
        );
        assert!(f.contains_key("poster_url") && f.contains_key("description/eng"));
        let cid = meta_feeder_sdk::hash::compute_yt_video_cid("video", "4D7u5KF7SP8").unwrap();
        assert_eq!(
            f.get(&format!("cids/{cid}")).map(String::as_str),
            Some("true")
        );

        // The query's anchor is echoed…
        assert_eq!(f.get("tmdbid").map(String::as_str), Some("95479"));
        // …but never promoted into a trust claim. A raw hit that stamped
        // `anchored` would leak into the enriched grid as if it had been
        // verified, which is the whole failure this tier is shaped to avoid.
        assert!(!f.contains_key("anchored"));
        // No guessed contentKind — `pack` in particular would be dropped
        // outright by meta-watch's raw projection.
        assert!(!f.contains_key("contentKind"));
    }

    #[tokio::test]
    async fn a_music_video_query_never_reaches_the_raw_path() {
        // meta-listen's artist row is `mbArtistId:<uuid> fileType:video
        // contentKind:musicVideo`. It requests video and would satisfy the raw
        // branch's file-type test, so the ONLY things keeping it on the
        // verified path are the dispatch order and the anchor guard. With no
        // MusicBrainz client configured the verified path yields nothing —
        // which is exactly how we can tell the two apart: a raw answer here
        // would be non-empty.
        let p = YouTubePlugin::default();
        let q = vq(
            "daft punk",
            &[
                ("mbArtistId", &["056e4f3e-d505-4dad-8ec1-d04f521cbb56"]),
                ("fileType", &["video"]),
                ("contentKind", &["musicVideo"]),
            ],
        );
        assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
    }

    #[tokio::test]
    async fn an_album_query_never_reaches_the_raw_path() {
        // meta-listen's album query is `fileType:audio` — it must keep falling
        // through to the MusicBrainz track path.
        let p = YouTubePlugin::default();
        let q = vq(
            "daft punk discovery",
            &[
                (
                    "mbReleaseGroupId",
                    &["aa997ea0-2936-40bd-884d-3af8a0e064dc"],
                ),
                ("fileType", &["audio"]),
                ("contentKind", &["track"]),
            ],
        );
        assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
    }

    #[tokio::test]
    async fn a_typed_video_query_with_no_free_text_returns_nothing() {
        // A `tmdbid` alone cannot drive the raw path: there is no TMDB client
        // here to turn the id into a searchable title, and searching for the
        // literal number would be worse than refusing.
        let p = YouTubePlugin::default();
        let q = vq("", &[("fileType", &["video"]), ("tmdbid", &["95479"])]);
        assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
    }

    #[tokio::test]
    async fn a_typed_video_query_answers_from_a_free_text_search() {
        use wiremock::matchers::{method, path, query_param};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/search"))
            .and(query_param("filter", "videos"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "items": [
                    {
                        "url": "/watch?v=4D7u5KF7SP8",
                        "title": "Three-Body EP01",
                        "uploaderUrl": "/channel/UCabc",
                        "uploaderName": "Tencent Video",
                        "duration": 2712
                    },
                    {
                        "url": "/watch?v=OLAK5uy_kZ8X",
                        "title": "Three-Body EP02",
                        "uploaderUrl": "/channel/UCabc",
                        "uploaderName": "Tencent Video",
                        "duration": 2680
                    }
                ]
            })))
            .mount(&server)
            .await;

        let p = YouTubePlugin::with_bases(server.uri(), server.uri());
        let q = vq("three-body", &[("fileType", &["video"])]);
        let out = p.handle_query(&q, 20).await.expect("ok");

        assert_eq!(out.len(), 2, "both hits emitted, unverified");
        assert_eq!(out[0].upstream_id, "youtube");
        assert_eq!(out[0].record_id, video_record_id("4D7u5KF7SP8"));
        assert_eq!(
            out[0].fields.get("ytMatchKey").map(String::as_str),
            Some(RAW_MATCH_KEY)
        );
        // No channel/duration filter ran — this is the point of the tier, and a
        // regression that reintroduced one would silently empty it.
        assert_eq!(
            out[1].fields.get("title").map(String::as_str),
            Some("Three-Body EP02")
        );
    }

    #[tokio::test]
    async fn the_raw_ceiling_bounds_one_query() {
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        // 30 distinct hits, an 11-char id each.
        let items: Vec<_> = (0..30)
            .map(|i| {
                serde_json::json!({
                    "url": format!("/watch?v=vid{:08}", i),
                    "title": format!("hit {i}"),
                    "uploaderUrl": "/channel/UCabc",
                    "uploaderName": "Someone",
                    "duration": 100
                })
            })
            .collect();
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/search"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "items": items
            })))
            .mount(&server)
            .await;

        let p = YouTubePlugin::with_bases(server.uri(), server.uri());
        let q = vq("anything", &[("fileType", &["video"])]);
        // `max_results` above the ceiling: the ceiling wins, because every hit
        // costs the core a thumbnail fetch.
        let out = p.handle_query(&q, 500).await.expect("ok");
        assert_eq!(out.len(), YT_VIDEO_SEARCH_MAX);

        // …and below it, the caller's own budget wins.
        let out = p.handle_query(&q, 3).await.expect("ok");
        assert_eq!(out.len(), 3);
    }

    fn tref(video: &str, delta: i64, key: &str) -> TrackRef {
        TrackRef {
            video_id: video.to_string(),
            title: video.to_string(),
            match_key: key.to_string(),
            delta,
        }
    }

    fn pick(recording: &str, refs: Vec<TrackRef>) -> TrackPick {
        TrackPick {
            recording_mbid: recording.to_string(),
            track: musicbrainz_feeder::client::Track {
                title: recording.to_string(),
                ..Default::default()
            },
            refs,
        }
    }

    /// §7.5.1 — the collision rule, with the study's own measured case.
    ///
    /// Burial's *Archangel* and *Near Dark* both resolved to `VVa_aCSA9Jc` at
    /// EXACT: same channel, both durations inside tolerance. They are different
    /// recordings, so at least one is wrong. The tighter delta keeps it and the
    /// other drops the reference entirely — emitting both would put the same
    /// audio under two names.
    #[test]
    fn two_recordings_never_share_one_video() {
        let mut picks = vec![
            pick("archangel", vec![tref("VVa_aCSA9Jc", 7, "mb:youtube-music")]),
            pick("near-dark", vec![tref("VVa_aCSA9Jc", 1, "mb:youtube-music")]),
        ];
        assign_references(&mut picks);
        assert!(picks[0].refs.is_empty(), "the looser claim loses the video");
        assert_eq!(picks[1].refs.len(), 1, "the tighter delta keeps it");
    }

    /// A collision must not cost a track its *other*, uncontested references.
    #[test]
    fn a_collision_only_drops_the_contested_reference() {
        let mut picks = vec![
            pick("a", vec![tref("shared", 9, "mb:youtube"), tref("own-a", 2, "mb:youtube")]),
            pick("b", vec![tref("shared", 1, "mb:youtube")]),
        ];
        assign_references(&mut picks);
        assert_eq!(picks[0].refs.len(), 1);
        assert_eq!(picks[0].refs[0].video_id, "own-a");
        assert_eq!(picks[1].refs[0].video_id, "shared");
    }

    /// §3 — several ranked references per recording is the rot mitigation.
    /// Uncontested ones all survive, in rank order.
    #[test]
    fn a_recording_keeps_several_ranked_references() {
        let mut picks = vec![pick(
            "rec",
            vec![
                tref("best", 0, "mb:youtube-music"),
                tref("second", 3, "mb:youtube"),
                tref("third", 8, "piped:music-artists"),
            ],
        )];
        assign_references(&mut picks);
        assert_eq!(picks[0].refs.len(), 3);
        assert_eq!(picks[0].refs[0].video_id, "best");
    }

    /// The tie-break behind that order: an equal duration delta is settled by
    /// how authoritative the channel key is, never arbitrarily.
    #[test]
    fn channel_key_strength_breaks_a_duration_tie() {
        assert!(key_priority("mb:youtube-music") < key_priority("mb:youtube"));
        assert!(key_priority("mb:youtube") < key_priority("piped:music-artists"));
    }

    #[test]
    fn the_cover_url_is_hqdefault_not_maxres() {
        // `maxresdefault` 404s on anything not uploaded in HD, and that 404 is
        // SILENT: the gate drops the record with no error anywhere.
        let url = format!("{YT_THUMB_BASE}/4D7u5KF7SP8/hqdefault.jpg");
        assert!(url.ends_with("/hqdefault.jpg"));
        assert!(meta_feeder_sdk::hash::artwork_locator(&url).is_some());
    }
}