use std::{
    ffi::OsStr,
    io::{self, Read},
    path::Path,
    process::{Child, Command, Stdio},
    sync::mpsc::{self, Receiver, TryRecvError},
    thread,
};

use serde::{Deserialize, Serialize};

pub const PAGE_SIZE: usize = 20;

pub struct Page {
    pub videos: Vec<Video>,
    pub has_more: bool,
    pub cutoff: Option<i64>,
}

#[derive(Clone, Deserialize, Serialize)]
pub struct Video {
    pub id: String,
    pub title: String,
    pub channel: Option<String>,
    pub channel_id: Option<String>,
    pub uploader_id: Option<String>,
    pub timestamp: Option<i64>,
    /// Cached YYYY-MM-DD label; a leading ~ marks an estimated UTC date.
    #[serde(default, rename = "upload_date")]
    pub publication_date: Option<String>,
}

#[derive(Deserialize)]
struct SearchResults<'a> {
    // Borrow raw rows so one schema error cannot abort siblings or copy unused metadata.
    #[serde(borrow)]
    entries: Vec<&'a serde_json::value::RawValue>,
    channel: Option<String>,
    channel_id: Option<String>,
    uploader_id: Option<String>,
}

pub fn search_results(output: &str) -> Result<Page, String> {
    let results: SearchResults = serde_json::from_str(output)
        .map_err(|error| format!("Invalid search response: {error}"))?;
    let has_more = results.entries.len() >= PAGE_SIZE;
    let videos = results
        .entries
        .into_iter()
        .filter_map(|entry| {
            let mut video = serde_json::from_str(entry.get()).ok()?;
            validate_video(&mut video).ok()?;
            Some(video)
        })
        .collect();
    Ok(Page {
        videos,
        has_more,
        cutoff: None,
    })
}

fn valid_channel_id(id: &str) -> bool {
    id.len() == 24
        && id.starts_with("UC")
        && id
            .bytes()
            .all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
}

pub(crate) fn valid_video_id(id: &str) -> bool {
    id.len() == 11
        && id
            .bytes()
            .all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
}

pub(crate) fn validate_video(video: &mut Video) -> Result<(), String> {
    if !valid_video_id(&video.id) {
        return Err("an invalid YouTube video ID".into());
    }
    if video
        .channel_id
        .as_deref()
        .is_some_and(|id| !valid_channel_id(id))
    {
        return Err("an invalid YouTube channel ID".into());
    }
    if video
        .uploader_id
        .as_deref()
        .is_some_and(|handle| handle.chars().any(char::is_control))
    {
        return Err("an invalid YouTube channel handle".into());
    }
    video.title.retain(|c| !c.is_control());
    if let Some(channel) = &mut video.channel {
        channel.retain(|c| !c.is_control());
    }
    let exact_date = video.publication_date.as_deref().and_then(|date| {
        if date.len() != 8 || !date.bytes().all(|byte| byte.is_ascii_digit()) {
            return None;
        }
        time::Date::from_calendar_date(
            date[..4].parse().ok()?,
            date[4..6].parse::<u8>().ok()?.try_into().ok()?,
            date[6..].parse().ok()?,
        )
        .ok()
    });
    video.publication_date = exact_date.map(|date| date.to_string()).or_else(|| {
        video.timestamp.and_then(|timestamp| {
            time::OffsetDateTime::from_unix_timestamp(timestamp)
                .ok()
                .map(|date| format!("~{}", date.date()))
        })
    });
    Ok(())
}

pub fn feed_results(output: &str) -> Result<Page, String> {
    let mut videos = Vec::new();
    let mut playlists = 0;
    let mut has_more = false;
    let mut cutoff = None;
    let mut unknown_frontier = false;
    // Multiple yt-dlp URLs produce one playlist object per line, not one combined object.
    for result in serde_json::Deserializer::from_str(output).into_iter::<SearchResults>() {
        let playlist = result.map_err(|error| format!("Invalid feed response: {error}"))?;
        playlists += 1;
        if playlist
            .channel_id
            .as_deref()
            .is_some_and(|id| !valid_channel_id(id))
        {
            return Err("Feed returned an invalid YouTube channel ID".into());
        }
        // Raw rows include unavailable videos; filtering them must not end pagination early.
        let full_page = playlist.entries.len() >= PAGE_SIZE;
        let first_video = videos.len();
        for entry in playlist.entries {
            let Ok(mut video) = serde_json::from_str::<Video>(entry.get()) else {
                continue;
            };
            if video.channel_id.is_none() {
                video.channel_id = playlist.channel_id.clone();
            }
            if video.uploader_id.is_none() {
                video.uploader_id = playlist.uploader_id.clone();
            }
            if video.channel.is_none() {
                video.channel = playlist.channel.clone();
            }
            if validate_video(&mut video).is_ok() {
                videos.push(video);
            }
        }
        if full_page {
            has_more = true;
            let oldest = videos[first_video..]
                .iter()
                .filter_map(|video| video.timestamp)
                .min();
            // A full undated/unavailable page may hide newer uploads on its next page.
            unknown_frontier |= oldest.is_none();
            cutoff = cutoff.max(oldest);
        }
    }
    if playlists == 0 {
        return Err("Feed returned no playlist data".into());
    }
    // Keep the newest occurrence of each ID without copying IDs into a second collection.
    videos.sort_unstable_by(|a, b| a.id.cmp(&b.id).then_with(|| b.timestamp.cmp(&a.timestamp)));
    videos.dedup_by(|a, b| a.id == b.id);
    videos.sort_by(|a, b| b.timestamp.cmp(&a.timestamp));
    Ok(Page {
        videos,
        has_more,
        cutoff: if unknown_frontier { None } else { cutoff },
    })
}

pub fn full_description(output: &str, expected_id: &str) -> Result<String, String> {
    #[derive(Deserialize)]
    struct Details {
        id: String,
        description: Option<String>,
    }
    let details: Details =
        serde_json::from_str(output).map_err(|error| format!("Invalid video metadata: {error}"))?;
    if details.id != expected_id {
        return Err("Video metadata did not match the requested video".into());
    }
    let mut description = details.description.unwrap_or_default();
    description.retain(|c| !c.is_control() || c == '\n' || c == '\t');
    Ok(description)
}

pub fn channel_handle(output: &str, expected_id: &str) -> Result<String, String> {
    #[derive(Deserialize)]
    struct Channel {
        channel_id: String,
        uploader_id: Option<String>,
    }
    let channel: Channel = serde_json::from_str(output)
        .map_err(|error| format!("Invalid channel metadata: {error}"))?;
    if channel.channel_id != expected_id {
        return Err("Channel metadata did not match the requested channel".into());
    }
    let handle = channel
        .uploader_id
        .filter(|handle| handle.starts_with('@'))
        .ok_or_else(|| "This channel has no available handle; no channel was saved.".to_owned())?;
    crate::config::normalize_channel(&handle).map_err(|error| error.to_string())?;
    Ok(handle)
}

fn page_end(start: usize) -> io::Result<usize> {
    start
        .checked_add(PAGE_SIZE - 1)
        .filter(|_| start != 0)
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid playlist page offset"))
}

// Own the extractor so replacing a request or quitting also cancels its process.
pub struct Job {
    child: Child,
    output: Receiver<io::Result<(String, String)>>,
}

impl Job {
    pub fn search(query: &str, start: usize) -> io::Result<Self> {
        let end = page_end(start)?;
        Self::spawn(extractor().args([
            "--flat-playlist",
            "--playlist-start",
            &start.to_string(),
            "--playlist-end",
            &end.to_string(),
            "--extractor-args",
            "youtubetab:approximate_date",
            "--dump-single-json",
            &format!("ytsearchall:{query}"),
        ]))
    }

    pub fn feed(channels: &[String], start: usize) -> io::Result<Self> {
        if channels.is_empty() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "No channels in this category",
            ));
        }
        let end = page_end(start)?;
        // ponytail: offsets traverse prior continuations; use direct tokens/per-channel exhaustion
        // tracking if deep archives become slow.
        Self::spawn(
            extractor()
                .args([
                    "--flat-playlist",
                    "--playlist-start",
                    &start.to_string(),
                    "--playlist-end",
                    &end.to_string(),
                    "--extractor-args",
                    "youtubetab:approximate_date",
                    "--dump-single-json",
                ])
                .args(channels.iter().map(|channel| format!("{channel}/videos"))),
        )
    }

    pub fn channel(id: &str) -> io::Result<Self> {
        if !valid_channel_id(id) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Invalid YouTube channel ID",
            ));
        }
        Self::spawn(extractor().args([
            "--flat-playlist",
            "--playlist-end",
            "1",
            "--dump-single-json",
            &format!("https://www.youtube.com/channel/{id}/videos"),
        ]))
    }

    pub fn description(id: &str) -> io::Result<Self> {
        Self::spawn(extractor().args([
            "--no-playlist",
            "--skip-download",
            "--dump-single-json",
            &format!("https://www.youtube.com/watch?v={id}"),
        ]))
    }

    fn spawn(command: &mut Command) -> io::Result<Self> {
        let mut child = command
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .map_err(|error| {
                io::Error::new(error.kind(), format!("Cannot start yt-dlp: {error}"))
            })?;
        let mut stdout = child.stdout.take().expect("stdout is piped");
        let mut stderr = child.stderr.take().expect("stderr is piped");
        let (sender, output) = mpsc::channel();
        // Drain both pipes while the process runs; a full pipe must not stall the TUI.
        thread::spawn(move || {
            let stdout_reader = thread::spawn(move || {
                let mut text = String::new();
                stdout.read_to_string(&mut text).map(|_| text)
            });
            let mut errors = String::new();
            let stderr_result = stderr.read_to_string(&mut errors);
            let result = stdout_reader
                .join()
                .unwrap_or_else(|_| Err(io::Error::other("Output reader panicked")))
                .and_then(|text| stderr_result.map(|_| (text, errors)));
            let _ = sender.send(result);
        });
        Ok(Self { child, output })
    }

    pub fn poll(&mut self) -> Option<Result<String, String>> {
        let status = match self.child.try_wait() {
            Ok(Some(status)) => status,
            Ok(None) => return None,
            Err(error) => return Some(Err(format!("Cannot wait for yt-dlp: {error}"))),
        };
        match self.output.try_recv() {
            Ok(Ok((output, _))) if status.success() => Some(Ok(output)),
            Ok(Ok((_, errors))) => Some(Err(if errors.trim().is_empty() {
                format!("yt-dlp exited with {status}")
            } else {
                errors.trim().to_owned()
            })),
            Ok(Err(error)) => Some(Err(format!("Cannot read yt-dlp output: {error}"))),
            Err(TryRecvError::Empty) => None,
            Err(TryRecvError::Disconnected) => {
                Some(Err("yt-dlp output reader disconnected".into()))
            }
        }
    }
}

impl Drop for Job {
    fn drop(&mut self) {
        stop(&mut self.child);
    }
}

const MAX_THUMBNAIL_BYTES: usize = 2 * 1024 * 1024;

pub struct ThumbnailJob {
    child: Child,
    output: Receiver<io::Result<(Vec<u8>, String)>>,
}

impl ThumbnailJob {
    pub fn start(id: &str) -> io::Result<Self> {
        if !valid_video_id(id) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Invalid YouTube video ID",
            ));
        }
        let mut child = isolated_command(OsStr::new("curl"))
            .args([
                "--disable",
                "--silent",
                "--show-error",
                "--fail",
                "--location",
                "--proto",
                "=https",
                "--proto-redir",
                "=https",
                "--connect-timeout",
                "5",
                "--max-time",
                "15",
                "--max-filesize",
                "2097152", // 2 MiB; read_thumbnail also enforces the limit.
            ])
            .arg(format!("https://i.ytimg.com/vi/{id}/hqdefault.jpg"))
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .map_err(|error| io::Error::new(error.kind(), format!("Cannot start curl: {error}")))?;
        let stdout = child.stdout.take().expect("stdout is piped");
        let errors = read_errors(child.stderr.take().expect("stderr is piped"));
        let (sender, output) = mpsc::channel();
        thread::spawn(move || {
            let bytes = read_thumbnail(stdout);
            let result = errors
                .recv()
                .unwrap_or_else(|_| Err(io::Error::other("curl error reader disconnected")))
                .and_then(|errors| bytes.map(|bytes| (bytes, errors)));
            let _ = sender.send(result);
        });
        Ok(Self { child, output })
    }

    pub fn poll(&mut self) -> Option<Result<Vec<u8>, String>> {
        // EOF alone is not success: curl can write a partial body and then fail.
        let status = match self.child.try_wait() {
            Ok(Some(status)) => status,
            Ok(None) => return None,
            Err(error) => return Some(Err(format!("Cannot wait for curl: {error}"))),
        };
        match self.output.try_recv() {
            Ok(Ok((bytes, _))) if status.success() => Some(Ok(bytes)),
            Ok(Ok((_, errors))) => Some(Err(if errors.trim().is_empty() {
                format!("curl exited with {status}")
            } else {
                errors.trim().to_owned()
            })),
            Ok(Err(error)) => Some(Err(format!("Cannot read thumbnail: {error}"))),
            Err(TryRecvError::Empty) => None,
            Err(TryRecvError::Disconnected) => Some(Err("curl output reader disconnected".into())),
        }
    }
}

impl Drop for ThumbnailJob {
    fn drop(&mut self) {
        stop(&mut self.child);
    }
}

fn read_thumbnail(pipe: impl Read) -> io::Result<Vec<u8>> {
    let mut bytes = Vec::new();
    pipe.take((MAX_THUMBNAIL_BYTES + 1) as u64)
        .read_to_end(&mut bytes)?;
    if bytes.len() > MAX_THUMBNAIL_BYTES {
        return Err(io::Error::other("Thumbnail exceeds 2 MiB"));
    }
    Ok(bytes)
}

fn isolated_command(program: &OsStr) -> Command {
    let mut command = Command::new(program);
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        command.process_group(0);
    }
    command
}

fn extractor() -> Command {
    let mut command = isolated_command(OsStr::new("yt-dlp"));
    command.args(["--ignore-config", "--no-warnings", "--socket-timeout", "15"]);
    command
}

fn stop(child: &mut Child) {
    if matches!(child.try_wait(), Ok(None)) {
        // SAFETY: isolated_command gives this still-running child its own process group.
        // Cancel the whole process group without signaling the TUI.
        #[cfg(unix)]
        unsafe {
            libc::kill(-(child.id() as i32), libc::SIGKILL);
        }
        let _ = child.kill();
    }
    let _ = child.wait();
}

fn read_errors(mut pipe: impl Read + Send + 'static) -> Receiver<io::Result<String>> {
    let (sender, receiver) = mpsc::channel();
    thread::spawn(move || {
        let mut text = String::new();
        let _ = sender.send(pipe.read_to_string(&mut text).map(|_| text));
    });
    receiver
}

fn failure(errors: &Receiver<io::Result<String>>, fallback: String) -> Option<Result<(), String>> {
    match errors.try_recv() {
        Ok(Ok(text)) => Some(Err(if text.trim().is_empty() {
            fallback
        } else {
            text.trim().to_owned()
        })),
        Ok(Err(error)) => Some(Err(format!("{fallback}: {error}"))),
        Err(TryRecvError::Empty) => None,
        Err(TryRecvError::Disconnected) => Some(Err(fallback)),
    }
}

pub struct PlayerLaunch {
    player: Child,
    player_errors: Receiver<io::Result<String>>,
}

impl PlayerLaunch {
    pub fn start(id: &str, player: &OsStr, progress_file: &Path) -> io::Result<Self> {
        if !valid_video_id(id) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Invalid YouTube video ID",
            ));
        }
        crate::video_state::register_identity(progress_file, id)?;
        let mut player = isolated_command(player)
            .args([
                "--new-instance",
                "--url",
                &format!("https://www.youtube.com/watch?v={id}"),
            ])
            .arg("--progress-file")
            .arg(progress_file)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .map_err(|error| {
                io::Error::new(
                    error.kind(),
                    format!("Cannot start Terebi launcher: {error}"),
                )
            })?;
        let player_errors = read_errors(player.stderr.take().expect("stderr is piped"));
        Ok(Self {
            player,
            player_errors,
        })
    }

    pub fn poll(&mut self) -> Option<Result<(), String>> {
        match self.player.try_wait() {
            Ok(Some(status)) if status.success() => Some(Ok(())),
            Ok(Some(status)) => {
                failure(&self.player_errors, format!("Player exited with {status}"))
            }
            Ok(None) => None,
            Err(error) => Some(Err(format!("Cannot wait for player: {error}"))),
        }
    }
}

impl Drop for PlayerLaunch {
    fn drop(&mut self) {
        stop(&mut self.player);
    }
}

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

    #[test]
    fn player_launch_rejects_unsafe_video_ids_before_spawning() {
        for id in ["", "--help", "jNQXAC9IVRw&list=other", "日本語"] {
            assert!(matches!(
                PlayerLaunch::start(id, OsStr::new("terebi"), Path::new("progress.jsonl")),
                Err(error) if error.kind() == io::ErrorKind::InvalidInput
            ));
        }
    }

    #[test]
    fn publication_dates_prefer_exact_metadata_and_handle_utc_boundaries_and_invalid_dates() {
        for parse in [search_results, feed_results] {
            for (upload_date, timestamp, expected) in [
                (Some("20240229"), Some(0), Some("2024-02-29")),
                (Some("20000229"), None, Some("2000-02-29")),
                (Some("19000229"), None, None),
                (Some("20230229"), Some(0), Some("~1970-01-01")),
                (Some("20241301"), None, None),
                (Some("日本語"), None, None),
                (None, Some(-1), Some("~1969-12-31")),
                (None, Some(1_709_251_199), Some("~2024-02-29")),
                (None, Some(1_709_251_200), Some("~2024-03-01")),
                (None, Some(i64::MAX), None),
                (None, None, None),
            ] {
                let output = serde_json::json!({"entries":[{
                    "id": "jNQXAC9IVRw", "title": "Video",
                    "upload_date": upload_date, "timestamp": timestamp,
                }]})
                .to_string();
                let page = parse(&output).unwrap();
                assert_eq!(page.videos[0].publication_date.as_deref(), expected);
            }
        }
    }

    #[test]
    fn thumbnails_bound_binary_input_and_reject_unsafe_ids() {
        let bytes = [0xff, 0xd8, 0x00, 0xfe];
        assert_eq!(read_thumbnail(bytes.as_slice()).unwrap(), bytes);
        assert_eq!(
            read_thumbnail(io::repeat(0).take(MAX_THUMBNAIL_BYTES as u64))
                .unwrap()
                .len(),
            MAX_THUMBNAIL_BYTES
        );
        assert!(read_thumbnail(io::repeat(0)).is_err());
        for id in ["../secrets", "abcdefghij/", "abcdefghijé", "--bad"] {
            assert!(matches!(
                ThumbnailJob::start(id),
                Err(error) if error.kind() == io::ErrorKind::InvalidInput
            ));
        }
    }

    #[test]
    fn bad_video_entries_do_not_discard_valid_results() {
        for parse in [search_results, feed_results] {
            for bad in [
                r#"{"id":"UC4QobU6STFB0P71PMvOGN5A","title":"Channel result"}"#,
                r#"{"id":"--bad","title":"Unsafe ID"}"#,
                r#"{"id":"jNQXAC9IVRw","title":"Bad channel","channel_id":"malformed"}"#,
                r#"{"id":"jNQXAC9IVRw","title":"Bad handle","uploader_id":"@jawed\u001b"}"#,
                r#"{"id":42,"title":"Wrong ID type"}"#,
                r#"{"id":"jNQXAC9IVRw","title":null}"#,
                r#"{"id":"jNQXAC9IVRw","title":"Bad date type","timestamp":"yesterday"}"#,
                r#"{"id":"jNQXAC9IVRw","title":"Bad channel type","channel":[]}"#,
                r#"{"id":"jNQXAC9IVRw"}"#,
                "null",
                "42",
                "[]",
                "{}",
            ] {
                let output = format!(
                    r#"{{"entries":[{{"id":"jNQXAC9IVRw","title":"Before"}},{bad},{{"id":"rQ_J9WH6CGk","title":"After"}}]}}"#,
                );
                let page = parse(&output).unwrap_or_else(|error| {
                    panic!("bad entry {bad} discarded the entire page: {error}")
                });
                assert_eq!(
                    page.videos
                        .iter()
                        .map(|video| video.title.as_str())
                        .collect::<Vec<_>>(),
                    ["Before", "After"]
                );
            }
        }
    }

    #[test]
    fn discarded_entries_still_count_for_pagination_but_not_feed_frontiers() {
        let bad = serde_json::json!({"id":"invalid", "title":"Bad", "timestamp":-500});
        let mut entries = vec![bad; PAGE_SIZE];
        for parse in [search_results, feed_results] {
            let page = parse(&serde_json::json!({"entries":entries}).to_string()).unwrap();
            assert!(page.videos.is_empty());
            assert!(
                page.has_more,
                "an all-invalid full page must not stop pagination"
            );
            assert_eq!(page.cutoff, None);
        }
        entries[0] = serde_json::json!({"id":"jNQXAC9IVRw","title":"Valid","timestamp":100});
        let page = feed_results(&serde_json::json!({"entries":entries}).to_string()).unwrap();
        assert!(page.has_more);
        assert_eq!(
            page.cutoff,
            Some(100),
            "invalid dates must not release old feed rows"
        );
        assert_eq!(page.videos[0].id, "jNQXAC9IVRw");
    }

    #[test]
    fn search_sanitizes_display_metadata_and_rejects_invalid_envelopes() {
        let videos = search_results(
            r#"{"entries":[null,{"id":"rQ_J9WH6CGk","title":"Rust\u001b","channel":null,"uploader_id":"@RustVideos"}]}"#,
        )
        .unwrap()
        .videos;
        assert_eq!(videos[0].title, "Rust");
        assert_eq!(videos[0].channel, None);
        assert_eq!(videos[0].channel_id, None);
        assert_eq!(videos[0].uploader_id.as_deref(), Some("@RustVideos"));
        for output in [
            "not JSON",
            "{}",
            r#"{"entries":null}"#,
            r#"{"entries":{}}"#,
            r#"{"entries":[{"id":} ]}"#,
        ] {
            assert!(search_results(output).is_err());
        }
    }

    #[test]
    fn feeds_merge_newest_first_deduplicate_and_inherit_channel_identity() {
        let output = r#"{"channel_id":"UCaYhcUwRBNscFNUKTjgPFiA","uploader_id":"@RustVideos","channel":"Rust\u001b","entries":[null,{"id":"l-GaaivVJCE","title":"Older","timestamp":100},{"id":"jNQXAC9IVRw","title":"Unknown","timestamp":null}]}
{"channel_id":"UCaYhcUwRBNscFNUKTjgPFiA","uploader_id":"@RustVideos","channel":"Rust","entries":[{"id":"SLV9AJxnb1U","title":"Newest","timestamp":300},{"id":"l-GaaivVJCE","title":"Newer","timestamp":200},{"id":"rQ_J9WH6CGk","title":"No metadata","channel_id":null}]}"#;
        let page = feed_results(output).unwrap();
        assert!(!page.has_more);
        assert_eq!(page.cutoff, None);
        let videos = page.videos;
        assert_eq!(
            videos
                .iter()
                .map(|video| video.id.as_str())
                .collect::<Vec<_>>(),
            ["SLV9AJxnb1U", "l-GaaivVJCE", "jNQXAC9IVRw", "rQ_J9WH6CGk"]
        );
        assert_eq!(videos[1].title, "Newer");
        assert_eq!(videos[1].timestamp, Some(200));
        assert_eq!(videos[2].timestamp, None);
        assert_eq!(videos[2].channel.as_deref(), Some("Rust"));
        assert!(
            videos
                .iter()
                .all(|video| video.channel_id.as_deref() == Some("UCaYhcUwRBNscFNUKTjgPFiA"))
        );
        assert!(
            videos
                .iter()
                .all(|video| video.uploader_id.as_deref() == Some("@RustVideos"))
        );
        let unavailable = feed_results(
            r#"{"entries":[null,{"id":"jNQXAC9IVRw","title":"Unknown","channel_id":null}]}"#,
        )
        .unwrap()
        .videos;
        assert_eq!(unavailable[0].channel_id, None);
        assert!(feed_results(r#"{"channel_id":"malformed","entries":[]}"#).is_err());
        assert!(feed_results("").is_err());
        assert!(feed_results("{}").is_err());
        assert!(feed_results("{not JSON}").is_err());
        let exhausted = feed_results(r#"{"entries":[]}"#).unwrap();
        assert!(exhausted.videos.is_empty());
        assert!(!exhausted.has_more);
        assert_eq!(exhausted.cutoff, None);
    }

    #[test]
    fn page_offsets_reject_zero_and_overflow_before_spawning() {
        assert_eq!(page_end(1).unwrap(), PAGE_SIZE);
        assert_eq!(page_end(PAGE_SIZE + 1).unwrap(), PAGE_SIZE * 2);
        assert_eq!(page_end(usize::MAX - (PAGE_SIZE - 1)).unwrap(), usize::MAX);
        let channels = vec!["https://www.youtube.com/@RustVideos".to_owned()];
        for start in [0, usize::MAX, usize::MAX - (PAGE_SIZE - 2)] {
            assert!(matches!(
                Job::search("rust", start),
                Err(error) if error.kind() == io::ErrorKind::InvalidInput
            ));
            assert!(matches!(
                Job::feed(&channels, start),
                Err(error) if error.kind() == io::ErrorKind::InvalidInput
            ));
        }
        assert!(matches!(
            Job::feed(&[], 1),
            Err(error) if error.kind() == io::ErrorKind::InvalidInput
        ));
    }

    #[test]
    fn search_pagination_counts_raw_rows_and_preserves_relevance_order() {
        for count in [0, PAGE_SIZE - 1, PAGE_SIZE, PAGE_SIZE + 1] {
            let output = serde_json::json!({
                "entries": vec![serde_json::Value::Null; count]
            })
            .to_string();
            let page = search_results(&output).unwrap();
            assert_eq!(page.has_more, count >= PAGE_SIZE);
            assert!(page.videos.is_empty());
            assert_eq!(page.cutoff, None);
        }
        let mut entries = vec![serde_json::Value::Null; PAGE_SIZE];
        entries[0] = serde_json::json!({
            "id": "aaaaaaaaaaa", "title": "Relevant older video", "timestamp": 100
        });
        entries[PAGE_SIZE - 1] = serde_json::json!({
            "id": "bbbbbbbbbbb", "title": "Less relevant newer video", "timestamp": 300
        });
        let page = search_results(&serde_json::json!({"entries": entries}).to_string()).unwrap();
        assert!(page.has_more);
        assert_eq!(page.cutoff, None);
        assert_eq!(page.videos.len(), 2);
        assert_eq!(page.videos[0].id, "aaaaaaaaaaa");
        assert_eq!(page.videos[1].id, "bbbbbbbbbbb");
    }

    #[test]
    fn feed_pagination_counts_nulls_and_duplicates_before_filtering() {
        for count in [0, PAGE_SIZE - 1, PAGE_SIZE, PAGE_SIZE + 1] {
            let output = serde_json::json!({
                "entries": vec![serde_json::Value::Null; count]
            })
            .to_string();
            let page = feed_results(&output).unwrap();
            assert_eq!(page.has_more, count >= PAGE_SIZE);
            assert!(page.videos.is_empty());
            assert_eq!(page.cutoff, None);
        }
        let entry = serde_json::json!({
            "id": "aaaaaaaaaaa", "title": "Repeated", "timestamp": 123
        });
        let page =
            feed_results(&serde_json::json!({"entries": vec![entry; PAGE_SIZE]}).to_string())
                .unwrap();
        assert!(page.has_more);
        assert_eq!(page.cutoff, Some(123));
        assert_eq!(page.videos.len(), 1);

        let mut entries = vec![serde_json::Value::Null; PAGE_SIZE];
        entries[0] = serde_json::json!({"id": "bbbbbbbbbbb", "title": "Unknown"});
        let page = feed_results(&serde_json::json!({"entries": entries}).to_string()).unwrap();
        assert!(page.has_more);
        assert_eq!(page.cutoff, None);
        assert_eq!(page.videos.len(), 1);
        assert_eq!(page.videos[0].timestamp, None);
    }

    #[test]
    fn feed_frontier_uses_only_full_channel_minima_and_retains_all_rows() {
        let mut first = vec![serde_json::Value::Null; PAGE_SIZE];
        first[0] =
            serde_json::json!({"id":"aaaaaaaaaaa","title":"Older duplicate","timestamp":500});
        first[1] = serde_json::json!({"id":"bbbbbbbbbbb","title":"First frontier","timestamp":300});
        first[2] = serde_json::json!({"id":"ccccccccccc","title":"Unknown"});
        let mut second = vec![serde_json::Value::Null; PAGE_SIZE];
        second[0] =
            serde_json::json!({"id":"aaaaaaaaaaa","title":"Newest duplicate","timestamp":600});
        second[1] =
            serde_json::json!({"id":"ddddddddddd","title":"Second frontier","timestamp":400});
        let short = serde_json::json!({"entries":[
            {"id":"eeeeeeeeeee","title":"Short newer","timestamp":999}
        ]});
        let older_short = serde_json::json!({"entries":[
            {"id":"fffffffffff","title":"Short older","timestamp":50}
        ]});
        let output = format!(
            "{}\n{}\n{}\n{}\n{}",
            serde_json::json!({"entries": first}),
            serde_json::json!({"entries": second}),
            short,
            older_short,
            serde_json::json!({"entries":[]})
        );
        let page = feed_results(&output).unwrap();
        assert!(page.has_more);
        assert_eq!(page.cutoff, Some(400));
        assert_eq!(
            page.videos
                .iter()
                .map(|video| video.id.as_str())
                .collect::<Vec<_>>(),
            [
                "eeeeeeeeeee",
                "aaaaaaaaaaa",
                "ddddddddddd",
                "bbbbbbbbbbb",
                "fffffffffff",
                "ccccccccccc"
            ]
        );
        assert_eq!(page.videos[1].title, "Newest duplicate");
        assert_eq!(page.videos.last().unwrap().timestamp, None);

        let exhausted = feed_results(&format!(
            "{}\n{}\n{}",
            short,
            older_short,
            serde_json::json!({"entries":[]})
        ))
        .unwrap();
        assert!(!exhausted.has_more);
        assert_eq!(exhausted.cutoff, None);
        assert_eq!(exhausted.videos.len(), 2);
    }

    #[test]
    fn channel_handles_require_the_requested_identity_and_a_valid_handle() {
        let id = "UChAQQ9at2MDlXjvQX3vL3Yw";
        for handle in ["@ch.1676", "@敷嶋てとら", "@hontame_"] {
            let metadata = serde_json::json!({"channel_id": id, "uploader_id": handle}).to_string();
            assert_eq!(channel_handle(&metadata, id).unwrap(), handle);
            assert!(channel_handle(&metadata, "UCaYhcUwRBNscFNUKTjgPFiA").is_err());
        }
        for handle in [
            None,
            Some("ch.1676"),
            Some("@bad/handle"),
            Some("@bad\u{1b}"),
        ] {
            let metadata = serde_json::json!({"channel_id": id, "uploader_id": handle}).to_string();
            assert!(channel_handle(&metadata, id).is_err());
        }
        assert!(channel_handle("not JSON", id).is_err());
    }

    #[test]
    fn full_descriptions_preserve_paragraphs_and_reject_mismatched_videos() {
        let metadata = r#"{"id":"jNQXAC9IVRw","description":"First paragraph.\n\n日本語\tsecond paragraph.\u001b\nLast line."}"#;
        assert_eq!(
            full_description(metadata, "jNQXAC9IVRw").unwrap(),
            "First paragraph.\n\n日本語\tsecond paragraph.\nLast line."
        );
        assert!(full_description(metadata, "rQ_J9WH6CGk").is_err());
        assert!(full_description("not JSON", "jNQXAC9IVRw").is_err());
        assert_eq!(
            full_description(r#"{"id":"jNQXAC9IVRw","description":null}"#, "jNQXAC9IVRw").unwrap(),
            ""
        );
    }
}