use std::{
    collections::VecDeque,
    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 = 60;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FeedRequest {
    pub channel: usize,
    pub start: usize,
    pub count: usize,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FeedProgress {
    pub request: FeedRequest,
    /// Raw rows, including unavailable, invalid and duplicate entries.
    pub count: usize,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct FeedCursor {
    // None before planning; an exhausted channel has no next offset.
    positions: Option<Vec<Option<usize>>>,
    rotation: usize,
}

impl FeedCursor {
    /// Planning is repeatable until a complete successful page is accepted.
    pub fn plan(&mut self, channels: usize) -> io::Result<Vec<FeedRequest>> {
        let positions = self
            .positions
            .get_or_insert_with(|| vec![Some(1); channels]);
        if positions.len() != channels {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Feed channel membership changed without resetting its cursor",
            ));
        }
        self.planned().map(|(requests, _)| requests)
    }

    fn planned(&self) -> io::Result<(Vec<FeedRequest>, usize)> {
        let Some(positions) = &self.positions else {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Feed is not planned",
            ));
        };
        let active = positions
            .iter()
            .filter(|position| position.is_some())
            .count();
        if active == 0 {
            return Ok((Vec::new(), self.rotation));
        }
        let selected = active.min(PAGE_SIZE);
        let mut requests = Vec::with_capacity(selected);
        let mut next_rotation = self.rotation;
        for step in 0..positions.len() {
            let channel = (self.rotation + step) % positions.len();
            let Some(start) = positions[channel] else {
                continue;
            };
            let index = requests.len();
            let count = PAGE_SIZE / selected + usize::from(index < PAGE_SIZE % selected);
            range_end(start, count)?;
            requests.push(FeedRequest {
                channel,
                start,
                count,
            });
            if index == (PAGE_SIZE - 1) % selected {
                next_rotation = (channel + 1) % positions.len();
            }
            if requests.len() == selected {
                break;
            }
        }
        requests.sort_unstable_by_key(|request| (request.start, request.count, request.channel));
        Ok((requests, next_rotation))
    }

    pub fn accept(&mut self, progress: &[FeedProgress]) -> Result<(), String> {
        let (requests, rotation) = self.planned().map_err(|error| error.to_string())?;
        if progress.len() != requests.len() {
            return Err("Feed progress did not match the planned channels".into());
        }
        // Preflight every request and offset before changing any channel or the rotation.
        for (entry, request) in progress.iter().zip(&requests) {
            if entry.request != *request || entry.count > request.count {
                return Err("Feed progress did not match the planned request".into());
            }
            if entry.count == request.count && request.start.checked_add(request.count).is_none() {
                return Err("Feed channel offset overflow".into());
            }
        }
        let positions = self.positions.as_mut().unwrap();
        for entry in progress {
            positions[entry.request.channel] = if entry.count < entry.request.count {
                None
            } else {
                Some(entry.request.start + entry.request.count)
            };
        }
        self.rotation = rotation;
        Ok(())
    }

    /// Unrequested channels remain active until their own response proves exhaustion.
    pub fn has_more(&self) -> bool {
        self.positions
            .as_ref()
            .is_some_and(|positions| positions.iter().any(Option::is_some))
    }
}

pub struct Page {
    pub videos: Vec<Video>,
    pub has_more: bool,
    pub feed_progress: Vec<FeedProgress>,
}

#[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,
        feed_progress: Vec::new(),
    })
}

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, requests: &[FeedRequest]) -> Result<Page, String> {
    validate_feed_requests(requests).map_err(|error| error.to_string())?;
    let mut videos = Vec::new();
    let mut feed_progress = Vec::with_capacity(requests.len());
    let mut playlists = serde_json::Deserializer::from_str(output).into_iter::<SearchResults>();
    // Each URL produces one playlist envelope; grouped command output follows request order.
    for &request in requests {
        let playlist = playlists
            .next()
            .ok_or_else(|| "Feed returned fewer playlists than requested".to_owned())?
            .map_err(|error| format!("Invalid feed response: {error}"))?;
        if playlist
            .channel_id
            .as_deref()
            .is_some_and(|id| !valid_channel_id(id))
        {
            return Err("Feed returned an invalid YouTube channel ID".into());
        }
        let count = playlist.entries.len();
        if count > request.count {
            return Err("Feed returned more rows than requested".into());
        }
        feed_progress.push(FeedProgress { request, count });
        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 let Some(extra) = playlists.next() {
        extra.map_err(|error| format!("Invalid feed response: {error}"))?;
        return Err("Feed returned more playlists than requested".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: feed_progress
            .iter()
            .any(|entry| entry.count == entry.request.count),
        feed_progress,
    })
}

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> {
    range_end(start, PAGE_SIZE)
}

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

fn validate_feed_requests(requests: &[FeedRequest]) -> io::Result<()> {
    let mut remaining = PAGE_SIZE;
    if requests.is_empty() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "No channels to request",
        ));
    }
    for (index, request) in requests.iter().enumerate() {
        range_end(request.start, request.count)?;
        remaining = remaining.checked_sub(request.count).ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "Feed request exceeds 60 total rows",
            )
        })?;
        if requests[..index]
            .iter()
            .any(|prior| prior.channel == request.channel)
        {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Duplicate feed channel",
            ));
        }
    }
    Ok(())
}

const MAX_EXTRACTORS: usize = 4;

// Own running and queued extractors so replacing a request cancels the whole page.
pub struct Job {
    children: Vec<Extraction>,
    queued: VecDeque<Command>,
}

struct Extraction {
    child: Option<Child>,
    output: Receiver<io::Result<(String, String)>>,
    status: Option<std::process::ExitStatus>,
    completed: Option<String>,
}

fn encode_query(query: &str) -> String {
    const HEX: &[u8; 16] = b"0123456789ABCDEF";
    let unreserved =
        |byte: u8| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~');
    let capacity = query
        .bytes()
        .map(|byte| if unreserved(byte) { 1 } else { 3 })
        .sum();
    let mut encoded = String::with_capacity(capacity);
    for byte in query.bytes() {
        if unreserved(byte) {
            encoded.push(char::from(byte));
        } else {
            encoded.push('%');
            encoded.push(char::from(HEX[usize::from(byte >> 4)]));
            encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
        }
    }
    encoded
}

fn feed_commands(
    channels: &[String],
    requests: &[FeedRequest],
    query: &str,
) -> io::Result<Vec<Command>> {
    validate_feed_requests(requests)?;
    let query = (!query.is_empty()).then(|| encode_query(query));
    let mut commands = Vec::new();
    let mut bounds = None;
    let batch_size = requests.len().div_ceil(MAX_EXTRACTORS);
    let mut batch_len = 0;
    for (index, request) in requests.iter().enumerate() {
        let channel = channels.get(request.channel).ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "Feed channel index is out of range",
            )
        })?;
        if requests[..index]
            .iter()
            .any(|prior| &channels[prior.channel] == channel)
        {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Duplicate feed channel URL",
            ));
        }
        if bounds != Some((request.start, request.count)) || batch_len == batch_size {
            let mut command = extractor();
            command.args([
                "--flat-playlist",
                "--playlist-start",
                &request.start.to_string(),
                "--playlist-end",
                &range_end(request.start, request.count)?.to_string(),
                "--extractor-args",
                "youtubetab:approximate_date",
                "--dump-single-json",
            ]);
            commands.push(command);
            bounds = Some((request.start, request.count));
            batch_len = 0;
        }
        commands.last_mut().unwrap().arg(match &query {
            Some(query) => format!("{channel}/search?query={query}"),
            None => format!("{channel}/videos"),
        });
        batch_len += 1;
    }
    Ok(commands)
}

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], requests: &[FeedRequest], query: &str) -> io::Result<Self> {
        Self::spawn_all(feed_commands(channels, requests, query)?)
    }

    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> {
        Ok(Self {
            children: vec![Extraction::spawn(command)?],
            queued: VecDeque::new(),
        })
    }

    fn spawn_all(commands: impl IntoIterator<Item = Command>) -> io::Result<Self> {
        let mut job = Self {
            children: Vec::new(),
            queued: VecDeque::new(),
        };
        // Dropping job also cancels started children if any initial spawn fails.
        for mut command in commands {
            if job.children.len() < MAX_EXTRACTORS {
                job.children.push(Extraction::spawn(&mut command)?);
            } else {
                job.queued.push_back(command);
            }
        }
        Ok(job)
    }

    pub fn poll(&mut self) -> Option<Result<String, String>> {
        if self.children.is_empty() {
            return None;
        }
        let mut active = 0;
        for child in &mut self.children {
            if child.completed.is_some() {
                continue;
            }
            match child.poll() {
                None => active += 1,
                Some(Ok(output)) => child.completed = Some(output),
                Some(Err(error)) => {
                    self.queued.clear();
                    self.children.clear();
                    return Some(Err(error));
                }
            }
        }
        while active < MAX_EXTRACTORS {
            let Some(mut command) = self.queued.pop_front() else {
                break;
            };
            match Extraction::spawn(&mut command) {
                Ok(child) => {
                    self.children.push(child);
                    active += 1;
                }
                Err(error) => {
                    self.queued.clear();
                    self.children.clear();
                    return Some(Err(error.to_string()));
                }
            }
        }
        if active != 0 {
            return None;
        }
        let mut children = self.children.drain(..);
        // Retain command order even when a queued group completes ahead of an earlier one.
        let mut output = children.next()?.completed.take().unwrap();
        output.reserve(
            children
                .as_slice()
                .iter()
                .map(|child| child.completed.as_ref().unwrap().len() + 1)
                .sum(),
        );
        for mut child in children {
            let next = child.completed.take().unwrap();
            if !output.is_empty() && !next.is_empty() {
                output.push('\n');
            }
            output.push_str(&next);
        }
        Some(Ok(output))
    }
}

impl Extraction {
    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: Some(child),
            output,
            status: None,
            completed: None,
        })
    }

    fn poll(&mut self) -> Option<Result<String, String>> {
        if self.status.is_none() {
            self.status = match self.child.as_mut()?.try_wait() {
                Ok(status) => status,
                Err(error) => return Some(Err(format!("Cannot wait for yt-dlp: {error}"))),
            };
        }
        let status = self.status?;
        if !status.success() {
            // Do not wait for inherited pipes held open by a failed child's descendants.
            self.cancel();
        }
        let result = 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()))
            }
        };
        if result.is_some() {
            self.cancel();
        }
        result
    }

    fn cancel(&mut self) {
        if let Some(mut child) = self.child.take() {
            // The leader may have exited while descendants still own its output pipes.
            // SAFETY: every extraction owns an isolated process group.
            #[cfg(unix)]
            unsafe {
                libc::kill(-(child.id() as i32), libc::SIGKILL);
            }
            stop(&mut child);
        }
    }
}

impl Drop for Extraction {
    fn drop(&mut self) {
        self.cancel();
    }
}

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 fn open_in_browser(id: &str) -> io::Result<Receiver<Result<(), String>>> {
    if !valid_video_id(id) {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "Invalid YouTube video ID",
        ));
    }
    let opener = if cfg!(target_os = "macos") {
        "open"
    } else {
        "xdg-open"
    };
    let child = isolated_command(OsStr::new(opener))
        .arg(format!("https://www.youtube.com/watch?v={id}"))
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .map_err(|error| {
            io::Error::new(
                error.kind(),
                format!("Cannot start browser opener: {error}"),
            )
        })?;
    let (sender, receiver) = mpsc::channel();
    thread::spawn(move || {
        let result = child
            .wait_with_output()
            .map_err(|error| format!("Cannot wait for browser opener: {error}"))
            .and_then(|output| {
                if output.status.success() {
                    Ok(())
                } else {
                    let error = String::from_utf8_lossy(&output.stderr);
                    Err(format!(
                        "Browser opener exited with {}: {}",
                        output.status,
                        error.trim()
                    ))
                }
            });
        let _ = sender.send(result);
    });
    Ok(receiver)
}

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 command = isolated_command(player);
        command
            .args([
                "--new-instance",
                "--url",
                &format!("https://www.youtube.com/watch?v={id}"),
            ])
            .arg("--progress-file")
            .arg(progress_file);
        Self::launch(command)
    }

    /// Session files are owned and retained by video_state, not by this launcher.
    pub fn start_source(
        source_file: &Path,
        player: &OsStr,
        progress_file: &Path,
    ) -> io::Result<Self> {
        if !std::fs::symlink_metadata(source_file)?.is_file() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Source document must be a regular file",
            ));
        }
        let source_file = std::fs::canonicalize(source_file)?;
        let parent = std::fs::canonicalize(progress_file.parent().ok_or_else(|| {
            io::Error::new(io::ErrorKind::InvalidInput, "Missing progress directory")
        })?)?;
        if source_file.parent() != Some(parent.as_path())
            || progress_file
                .file_name()
                .is_none_or(|name| name != "progress.jsonl")
        {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Source and progress must share their private session directory",
            ));
        }
        let progress_file = parent.join("progress.jsonl");
        match std::fs::symlink_metadata(&progress_file) {
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
            Ok(_) => {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    "Progress destination must not exist",
                ));
            }
        }
        let mut command = isolated_command(player);
        command
            .arg("--new-instance")
            .arg("--source-file")
            .arg(source_file)
            .arg("--progress-file")
            .arg(progress_file);
        Self::launch(command)
    }

    fn launch(mut command: Command) -> io::Result<Self> {
        let mut player = command
            .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::*;

    const ONE_CHANNEL: [FeedRequest; 1] = [FeedRequest {
        channel: 0,
        start: 1,
        count: PAGE_SIZE,
    }];

    fn full_progress(requests: &[FeedRequest]) -> Vec<FeedProgress> {
        requests
            .iter()
            .map(|&request| FeedProgress {
                request,
                count: request.count,
            })
            .collect()
    }

    #[test]
    fn feed_cursor_shares_60_rows_across_17_channels_with_continuous_uneven_offsets() {
        let mut cursor = FeedCursor::default();
        assert!(!cursor.has_more());
        let mut offsets = [1; 17];
        for _ in 0..3 {
            let requests = cursor.plan(17).unwrap();
            let planned = cursor.clone();
            assert_eq!(
                cursor.plan(17).unwrap(),
                requests,
                "retrying must preserve the plan"
            );
            assert_eq!(
                cursor, planned,
                "planning must not advance offsets or rotation"
            );
            assert_eq!(requests.len(), 17);
            assert_eq!(
                requests.iter().map(|request| request.count).sum::<usize>(),
                PAGE_SIZE
            );
            assert_eq!(
                requests.iter().filter(|request| request.count == 4).count(),
                9
            );
            assert_eq!(
                requests.iter().filter(|request| request.count == 3).count(),
                8
            );
            assert!(requests.windows(2).all(|pair| {
                (pair[0].start, pair[0].count, pair[0].channel)
                    < (pair[1].start, pair[1].count, pair[1].channel)
            }));
            for request in &requests {
                assert_eq!(request.start, offsets[request.channel]);
                offsets[request.channel] += request.count;
            }
            cursor.accept(&full_progress(&requests)).unwrap();
        }
        // Rotating the nine remainder rows avoids permanently favoring the first channels.
        assert!(offsets.iter().max().unwrap() - offsets.iter().min().unwrap() <= 1);
        let saved = cursor.clone();
        assert!(cursor.plan(16).is_err());
        assert_eq!(cursor, saved);
    }

    #[test]
    fn exhausted_channel_quota_moves_to_survivors_without_restarting_offsets() {
        let mut cursor = FeedCursor::default();
        let first = cursor.plan(3).unwrap();
        let mut progress = full_progress(&first);
        progress
            .iter_mut()
            .find(|entry| entry.request.channel == 1)
            .unwrap()
            .count = 0;
        cursor.accept(&progress).unwrap();
        let second = cursor.plan(3).unwrap();
        assert_eq!(
            second,
            [
                FeedRequest {
                    channel: 0,
                    start: 21,
                    count: 30
                },
                FeedRequest {
                    channel: 2,
                    start: 21,
                    count: 30
                },
            ]
        );
        let mut progress = full_progress(&second);
        progress[0].count = 29;
        cursor.accept(&progress).unwrap();
        let last = cursor.plan(3).unwrap();
        assert_eq!(
            last,
            [FeedRequest {
                channel: 2,
                start: 51,
                count: PAGE_SIZE
            }]
        );
        cursor
            .accept(&[FeedProgress {
                request: last[0],
                count: 0,
            }])
            .unwrap();
        assert!(!cursor.has_more());
        assert!(cursor.plan(3).unwrap().is_empty());
        let mut empty = FeedCursor::default();
        assert!(empty.plan(0).unwrap().is_empty());
        assert!(!empty.has_more());
    }

    #[test]
    fn more_than_60_channels_rotate_fairly_and_unvisited_channels_do_not_exhaust() {
        let mut cursor = FeedCursor::default();
        let mut seen = [false; 137];
        let mut offsets = vec![1; 137];
        for _ in 0..3 {
            let requests = cursor.plan(137).unwrap();
            assert_eq!(requests.len(), PAGE_SIZE);
            for request in &requests {
                assert_eq!(request.count, 1);
                assert_eq!(request.start, offsets[request.channel]);
                seen[request.channel] = true;
                offsets[request.channel] += 1;
            }
            cursor.accept(&full_progress(&requests)).unwrap();
        }
        assert!(seen.iter().all(|seen| *seen));
        let mut cursor = FeedCursor::default();
        for expected_remaining in [77, 17, 0] {
            let requests = cursor.plan(137).unwrap();
            assert_eq!(
                requests.iter().map(|request| request.count).sum::<usize>(),
                PAGE_SIZE
            );
            let progress: Vec<_> = requests
                .iter()
                .map(|&request| FeedProgress { request, count: 0 })
                .collect();
            cursor.accept(&progress).unwrap();
            assert_eq!(
                cursor
                    .positions
                    .as_ref()
                    .unwrap()
                    .iter()
                    .filter(|offset| offset.is_some())
                    .count(),
                expected_remaining,
            );
            assert_eq!(cursor.has_more(), expected_remaining != 0);
        }
        assert!(cursor.plan(137).unwrap().is_empty());
    }

    #[test]
    fn cursor_rejects_invalid_or_stale_progress_and_overflow_atomically() {
        let mut cursor = FeedCursor::default();
        assert!(cursor.accept(&[]).is_err());
        let requests = cursor.plan(3).unwrap();
        let valid = full_progress(&requests);
        let saved = cursor.clone();
        let mut bad = valid.clone();
        bad[2].count += 1;
        assert!(cursor.accept(&bad).is_err());
        assert_eq!(cursor, saved);
        bad = valid.clone();
        bad[2].request.channel = 99;
        assert!(cursor.accept(&bad).is_err());
        assert_eq!(cursor, saved);
        bad = valid.clone();
        bad[2].request = bad[0].request;
        assert!(cursor.accept(&bad).is_err());
        assert_eq!(cursor, saved);
        bad = valid.clone();
        bad[2].request.start += 1;
        assert!(cursor.accept(&bad).is_err());
        assert_eq!(cursor, saved);
        bad = valid.clone();
        bad.reverse();
        assert!(cursor.accept(&bad).is_err());
        assert_eq!(cursor, saved);
        assert!(cursor.accept(&valid[..2]).is_err());
        assert_eq!(cursor, saved);
        cursor.accept(&valid).unwrap();
        let accepted = cursor.clone();
        assert!(cursor.accept(&valid).is_err());
        assert_eq!(cursor, accepted);

        let mut overflow = FeedCursor {
            positions: Some(vec![Some(1), Some(usize::MAX - 19), Some(1)]),
            rotation: 0,
        };
        let requests = overflow.plan(3).unwrap();
        let saved = overflow.clone();
        assert!(overflow.accept(&full_progress(&requests)).is_err());
        assert_eq!(overflow, saved);
    }

    #[cfg(unix)]
    mod subprocesses {
        use super::*;

        fn controlled(script: &str) -> Command {
            let mut command = isolated_command(OsStr::new("/bin/sh"));
            command.args(["-c", &format!("kill -STOP $$; {script}")]);
            command
        }

        fn wait_stopped(extraction: &Extraction) {
            wait_stopped_pid(extraction.child.as_ref().unwrap().id());
        }

        fn wait_stopped_pid(pid: u32) {
            let pid = pid as i32;
            let mut status = 0;
            loop {
                // SAFETY: wait only for this test's owned child, without reaping its exit.
                let result = unsafe { libc::waitpid(pid, &mut status, libc::WUNTRACED) };
                if result == -1 && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
                    continue;
                }
                assert_eq!(result, pid);
                assert!(libc::WIFSTOPPED(status));
                break;
            }
        }

        fn release(extraction: &mut Extraction) {
            let child = extraction.child.as_mut().unwrap();
            // SAFETY: this child is stopped at the deterministic test barrier.
            assert_eq!(unsafe { libc::kill(child.id() as i32, libc::SIGCONT) }, 0);
            child.wait().unwrap();
            // Synchronize pipe draining too, then leave the real result for Job::poll.
            // Blocking is confined to tests; production polling never waits for output.
            let result = extraction.output.recv().unwrap();
            let (sender, receiver) = mpsc::channel();
            sender.send(result).unwrap();
            extraction.output = receiver;
        }

        fn take_output(extraction: &mut Extraction) -> Receiver<io::Result<(String, String)>> {
            let (_, receiver) = mpsc::channel();
            std::mem::replace(&mut extraction.output, receiver)
        }

        fn assert_reaped(pid: u32) {
            let mut status = 0;
            // SAFETY: query only this test's child; ECHILD proves cancellation reaped it.
            assert_eq!(
                unsafe { libc::waitpid(pid as i32, &mut status, libc::WNOHANG) },
                -1
            );
            assert_eq!(
                io::Error::last_os_error().raw_os_error(),
                Some(libc::ECHILD)
            );
        }

        #[test]
        fn a_later_spawn_failure_cancels_and_reaps_already_started_batches() {
            static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
            let directory = std::env::temp_dir().join(format!(
                "tubular-extractor-spawn-{}-{}",
                std::process::id(),
                NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
            ));
            std::fs::create_dir(&directory).unwrap();
            let ready = directory.join("ready");
            let mut first = isolated_command(OsStr::new("/bin/sh"));
            first
                .args(["-c", "printf '%s' $$ > \"$1\"; kill -STOP $$", "test"])
                .arg(&ready);
            let mut started_pid = None;
            let commands = std::iter::once(first).chain(std::iter::once_with(|| {
                // The file only communicates identity; SIGSTOP is the completion barrier.
                let pid = loop {
                    if let Ok(text) = std::fs::read_to_string(&ready)
                        && let Ok(pid) = text.parse::<u32>()
                    {
                        break pid;
                    }
                    thread::yield_now();
                };
                wait_stopped_pid(pid);
                started_pid = Some(pid);
                isolated_command(directory.join("missing-extractor").as_os_str())
            }));
            let error = Job::spawn_all(commands).err().unwrap();
            assert_eq!(error.kind(), io::ErrorKind::NotFound);
            assert_reaped(started_pid.unwrap());
            std::fs::remove_dir_all(directory).unwrap();
        }

        #[test]
        fn queued_range_groups_keep_four_workers_and_preserve_parser_request_order() {
            let requests: Vec<_> = (0..6)
                .map(|channel| FeedRequest {
                    channel,
                    start: channel + 1,
                    count: channel + 1,
                })
                .collect();
            let commands = requests.iter().map(|request| {
                let output =
                    serde_json::json!({"entries": vec![serde_json::Value::Null; request.count]});
                controlled(&format!("printf '%s' '{output}'"))
            });
            let mut job = Job::spawn_all(commands).unwrap();
            assert_eq!(job.children.len(), MAX_EXTRACTORS);
            assert_eq!(job.queued.len(), 2);
            for child in &job.children {
                wait_stopped(child);
            }
            release(&mut job.children[3]);
            assert!(job.poll().is_none());
            assert_eq!(job.queued.len(), 1);
            assert_eq!(
                job.children
                    .iter()
                    .filter(|child| child.completed.is_none())
                    .count(),
                MAX_EXTRACTORS
            );
            wait_stopped(&job.children[4]);
            release(&mut job.children[4]);
            release(&mut job.children[1]);
            assert!(job.poll().is_none());
            assert!(job.queued.is_empty());
            assert!(
                job.children
                    .iter()
                    .filter(|child| child.completed.is_none())
                    .count()
                    <= MAX_EXTRACTORS
            );
            wait_stopped(&job.children[5]);
            release(&mut job.children[5]);
            assert!(job.poll().is_none());
            release(&mut job.children[2]);
            assert!(job.poll().is_none());
            release(&mut job.children[0]);
            let output = job.poll().unwrap().unwrap();
            let page = feed_results(&output, &requests).unwrap();
            assert_eq!(page.feed_progress, full_progress(&requests));
            assert!(page.videos.is_empty());
            assert!(job.poll().is_none());
        }

        #[test]
        fn queued_spawn_failure_discards_outputs_and_cancels_running_and_queued_groups() {
            let mut job = Job::spawn_all([
                controlled("printf '%s' 'completed output'"),
                controlled("printf '%s' 'pending one'"),
                controlled("printf '%s' 'pending two'"),
                controlled("printf '%s' 'pending three'"),
                isolated_command(OsStr::new("/dev/null/tubular-missing-extractor")),
                controlled("printf '%s' 'must never start'"),
            ])
            .unwrap();
            for child in &job.children {
                wait_stopped(child);
            }
            let pids: Vec<_> = job.children[1..]
                .iter()
                .map(|child| child.child.as_ref().unwrap().id())
                .collect();
            let outputs: Vec<_> = job.children[1..].iter_mut().map(take_output).collect();
            release(&mut job.children[0]);
            assert!(
                job.poll()
                    .unwrap()
                    .unwrap_err()
                    .starts_with("Cannot start yt-dlp:")
            );
            assert!(job.children.is_empty());
            assert!(job.queued.is_empty());
            for output in outputs {
                assert!(output.recv().unwrap().is_ok());
            }
            for pid in pids {
                assert_reaped(pid);
            }
            assert!(job.poll().is_none());
        }

        #[test]
        fn aggregate_waits_for_all_batches_and_merges_in_request_not_completion_order() {
            let mut job = Job::spawn_all([
                controlled("printf '%s' '{\"batch\":1}'"),
                controlled("printf '%s' '{\"batch\":2}'"),
                controlled("printf '%s' '{\"batch\":3}'"),
            ])
            .unwrap();
            for child in &job.children {
                wait_stopped(child);
            }
            assert!(job.poll().is_none());
            release(&mut job.children[2]);
            assert!(
                job.poll().is_none(),
                "the last batch cannot complete a page"
            );
            release(&mut job.children[1]);
            assert!(job.poll().is_none(), "the first batch is still outstanding");
            release(&mut job.children[0]);
            let output = job.poll().unwrap().unwrap();
            let batches: Vec<_> = serde_json::Deserializer::from_str(&output)
                .into_iter::<serde_json::Value>()
                .map(Result::unwrap)
                .collect();
            assert_eq!(
                batches,
                [
                    serde_json::json!({"batch":1}),
                    serde_json::json!({"batch":2}),
                    serde_json::json!({"batch":3}),
                ]
            );
            assert!(job.poll().is_none());
        }

        #[test]
        fn a_failed_group_discards_completed_output_and_cancels_running_and_queued_siblings() {
            let mut job = Job::spawn_all([
                controlled("printf '%s' '{\"complete\":true}'"),
                controlled(
                    "printf '%s' '{\"partial\":true}'; printf '%s' 'batch failed' >&2; exit 7",
                ),
                controlled("/bin/sh -c 'kill -STOP $$' & wait"),
                controlled("printf '%s' 'pending sibling'"),
                controlled("printf '%s' 'queued sibling'"),
                controlled("printf '%s' 'must never start'"),
            ])
            .unwrap();
            for child in &job.children {
                wait_stopped(child);
            }
            release(&mut job.children[0]);
            assert!(job.poll().is_none());
            wait_stopped(&job.children[4]);
            assert_eq!(job.queued.len(), 1);
            let pids: Vec<_> = job.children[2..]
                .iter()
                .map(|child| child.child.as_ref().unwrap().id())
                .collect();
            let outputs: Vec<_> = job.children[2..].iter_mut().map(take_output).collect();
            release(&mut job.children[1]);
            assert_eq!(job.poll().unwrap().unwrap_err(), "batch failed");
            assert!(job.children.is_empty());
            assert!(job.queued.is_empty());
            for output in outputs {
                assert!(output.recv().unwrap().is_ok());
            }
            for pid in pids {
                assert_reaped(pid);
            }
            assert!(job.poll().is_none());
        }

        #[test]
        fn unreadable_batch_output_fails_the_whole_page_and_cancels_siblings() {
            let mut job = Job::spawn_all([
                controlled("printf '\\377'"),
                controlled("printf '%s' 'must not finish'"),
            ])
            .unwrap();
            for child in &job.children {
                wait_stopped(child);
            }
            let pending_pid = job.children[1].child.as_ref().unwrap().id();
            let pending_output = take_output(&mut job.children[1]);
            release(&mut job.children[0]);
            assert!(
                job.poll()
                    .unwrap()
                    .unwrap_err()
                    .starts_with("Cannot read yt-dlp output:")
            );
            assert!(pending_output.recv().unwrap().is_ok());
            assert_reaped(pending_pid);
        }

        #[test]
        fn dropping_a_page_cancels_and_reaps_running_groups_and_drops_queued_commands() {
            let mut job = Job::spawn_all([
                controlled("/bin/sh -c 'kill -STOP $$' & wait"),
                controlled("printf '%s' 'must not finish'"),
                controlled("printf '%s' 'pending two'"),
                controlled("printf '%s' 'pending three'"),
                controlled("printf '%s' 'queued one must never start'"),
                controlled("printf '%s' 'queued two must never start'"),
            ])
            .unwrap();
            assert_eq!(job.children.len(), MAX_EXTRACTORS);
            assert_eq!(job.queued.len(), 2);
            for child in &job.children {
                wait_stopped(child);
            }
            let pids: Vec<_> = job
                .children
                .iter()
                .map(|child| child.child.as_ref().unwrap().id())
                .collect();
            let outputs: Vec<_> = job.children.iter_mut().map(take_output).collect();
            drop(job);
            for output in outputs {
                assert!(output.recv().unwrap().is_ok());
            }
            for pid in pids {
                assert_reaped(pid);
            }
        }

        #[test]
        fn dropping_after_the_leader_exits_still_cancels_inherited_pipe_owners() {
            let mut job =
                Job::spawn_all([controlled("/bin/sh -c 'kill -STOP $$' & exit 0")]).unwrap();
            wait_stopped(&job.children[0]);
            let child = job.children[0].child.as_mut().unwrap();
            let pid = child.id();
            // SAFETY: resume this test's stopped leader, which exits without its descendant.
            assert_eq!(unsafe { libc::kill(pid as i32, libc::SIGCONT) }, 0);
            assert!(child.wait().unwrap().success());
            let output = take_output(&mut job.children[0]);
            drop(job);
            // EOF proves the descendant that inherited stdout/stderr was also canceled.
            assert!(output.recv().unwrap().is_ok());
            assert_reaped(pid);
        }
    }

    #[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 browser_open_rejects_unsafe_video_ids_before_spawning() {
        for id in ["", "--help", "jNQXAC9IVRw&list=other", "日本語"] {
            assert!(matches!(
                open_in_browser(id),
                Err(error) if error.kind() == io::ErrorKind::InvalidInput
            ));
        }
    }

    #[test]
    fn publication_dates_prefer_exact_metadata_and_handle_utc_boundaries_and_invalid_dates() {
        for feed in [false, true] {
            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 = if feed {
                    feed_results(&output, &ONE_CHANNEL)
                } else {
                    search_results(&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 feed in [false, true] {
            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 = if feed {
                    feed_results(&output, &ONE_CHANNEL)
                } else {
                    search_results(&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_raw_entries_advance_the_cursor_only_after_acceptance() {
        let bad = serde_json::json!({"id":"invalid", "title":"Bad", "timestamp":-500});
        let mut entries = vec![bad; PAGE_SIZE];
        let output = serde_json::json!({"entries": entries}).to_string();
        let search = search_results(&output).unwrap();
        assert!(search.videos.is_empty());
        assert!(search.has_more);
        assert!(search.feed_progress.is_empty());
        let mut cursor = FeedCursor::default();
        let requests = cursor.plan(1).unwrap();
        let saved = cursor.clone();
        let page = feed_results(&output, &requests).unwrap();
        assert!(page.videos.is_empty());
        assert!(page.has_more);
        assert_eq!(page.feed_progress, full_progress(&requests));
        assert_eq!(cursor, saved, "parsing alone must not advance the cursor");
        cursor.accept(&page.feed_progress).unwrap();
        assert_eq!(cursor.plan(1).unwrap()[0].start, PAGE_SIZE + 1);

        entries[0] = serde_json::json!({"id":"jNQXAC9IVRw","title":"Valid","timestamp":100});
        let page = feed_results(
            &serde_json::json!({"entries": entries}).to_string(),
            &cursor.plan(1).unwrap(),
        )
        .unwrap();
        assert!(page.has_more);
        assert_eq!(page.feed_progress[0].count, PAGE_SIZE);
        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 requests = FeedCursor::default().plan(2).unwrap();
        let page = feed_results(output, &requests).unwrap();
        assert!(!page.has_more);
        assert_eq!(
            page.feed_progress
                .iter()
                .map(|entry| entry.count)
                .collect::<Vec<_>>(),
            [3, 3]
        );
        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}]}"#,
            &ONE_CHANNEL,
        )
        .unwrap()
        .videos;
        assert_eq!(unavailable[0].channel_id, None);
        assert!(feed_results(r#"{"channel_id":"malformed","entries":[]}"#, &ONE_CHANNEL).is_err());
        assert!(feed_results("", &ONE_CHANNEL).is_err());
        assert!(feed_results("{}", &ONE_CHANNEL).is_err());
        assert!(feed_results("{not JSON}", &ONE_CHANNEL).is_err());
        let exhausted = feed_results(r#"{"entries":[]}"#, &ONE_CHANNEL).unwrap();
        assert!(exhausted.videos.is_empty());
        assert!(!exhausted.has_more);
        assert_eq!(exhausted.feed_progress[0].count, 0);
    }

    #[test]
    fn channel_search_query_preserves_utf8_and_isolates_url_parameters() {
        assert_eq!(
            encode_query("café 🦀 &next=1#?/+%20\t\n\0 AZaz09-._~"),
            "caf%C3%A9%20%F0%9F%A6%80%20%26next%3D1%23%3F%2F%2B%2520%09%0A%00%20AZaz09-._~"
        );
    }

    #[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, &[FeedRequest { start, ..ONE_CHANNEL[0] }], ""),
                Err(error) if error.kind() == io::ErrorKind::InvalidInput
            ));
        }
        assert!(matches!(
            Job::feed(&[], &ONE_CHANNEL, ""),
            Err(error) if error.kind() == io::ErrorKind::InvalidInput
        ));
    }

    #[test]
    fn feeds_reject_invalid_request_budgets_and_envelope_or_row_count_mismatches() {
        let channels = vec!["https://www.youtube.com/@RustVideos".to_owned(); 2];
        for requests in [
            Vec::new(),
            vec![FeedRequest {
                count: 0,
                ..ONE_CHANNEL[0]
            }],
            vec![FeedRequest {
                count: PAGE_SIZE + 1,
                ..ONE_CHANNEL[0]
            }],
            vec![FeedRequest {
                channel: 2,
                ..ONE_CHANNEL[0]
            }],
            vec![
                FeedRequest {
                    channel: 0,
                    start: 1,
                    count: 30,
                },
                FeedRequest {
                    channel: 0,
                    start: 2,
                    count: 30,
                },
            ],
            vec![
                FeedRequest {
                    channel: 0,
                    start: 1,
                    count: 30,
                },
                FeedRequest {
                    channel: 1,
                    start: 1,
                    count: 30,
                },
            ],
            vec![
                FeedRequest {
                    channel: 0,
                    start: 1,
                    count: 31,
                },
                FeedRequest {
                    channel: 1,
                    start: 1,
                    count: 30,
                },
            ],
        ] {
            assert!(matches!(
                Job::feed(&channels, &requests, ""),
                Err(error) if error.kind() == io::ErrorKind::InvalidInput
            ));
        }
        let requests = FeedCursor::default().plan(2).unwrap();
        assert!(feed_results(r#"{"entries":[]}"#, &requests).is_err());
        for output in [
            r#"{"entries":[]} {"entries":[]}"#,
            r#"{"entries":[]} garbage"#,
            r#"{"entries":[null]} {"entries":null}"#,
            r#"{"entries":null}"#,
            r#"{"entries":{}}"#,
            r#"{"entries":[{"id":}]}"#,
        ] {
            assert!(feed_results(output, &ONE_CHANNEL).is_err());
        }
        let request = [FeedRequest {
            channel: 0,
            start: 1,
            count: 2,
        }];
        assert!(feed_results(r#"{"entries":[null,null,null]}"#, &request).is_err());
        assert!(feed_results(r#"{"entries":[]}"#, &[]).is_err());
        assert!(
            feed_results(
                r#"{"entries":[]}"#,
                &[FeedRequest {
                    count: 0,
                    ..ONE_CHANNEL[0]
                }]
            )
            .is_err()
        );
        assert!(
            feed_results(
                r#"{"entries":[]}"#,
                &[FeedRequest {
                    count: PAGE_SIZE + 1,
                    ..ONE_CHANNEL[0]
                }]
            )
            .is_err()
        );
    }

    #[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!(page.feed_progress.is_empty());
        }
        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!(page.feed_progress.is_empty());
        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] {
            let output = serde_json::json!({
                "entries": vec![serde_json::Value::Null; count]
            })
            .to_string();
            let page = feed_results(&output, &ONE_CHANNEL).unwrap();
            assert_eq!(page.has_more, count == PAGE_SIZE);
            assert!(page.videos.is_empty());
            assert_eq!(
                page.feed_progress,
                [FeedProgress {
                    request: ONE_CHANNEL[0],
                    count
                }]
            );
        }
        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(),
            &ONE_CHANNEL,
        )
        .unwrap();
        assert!(page.has_more);
        assert_eq!(page.feed_progress[0].count, PAGE_SIZE);
        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(),
            &ONE_CHANNEL,
        )
        .unwrap();
        assert!(page.has_more);
        assert_eq!(page.feed_progress[0].count, PAGE_SIZE);
        assert_eq!(page.videos.len(), 1);
        assert_eq!(page.videos[0].timestamp, None);
    }

    #[test]
    fn feeds_retain_short_and_undated_rows_without_a_global_frontier() {
        let requests = FeedCursor::default().plan(5).unwrap();
        let mut first = vec![serde_json::Value::Null; requests[0].count];
        first[0] =
            serde_json::json!({"id":"aaaaaaaaaaa","title":"Older duplicate","timestamp":500});
        first[1] =
            serde_json::json!({"id":"bbbbbbbbbbb","title":"First dated row","timestamp":300});
        first[2] = serde_json::json!({"id":"ccccccccccc","title":"Unknown"});
        let mut second = vec![serde_json::Value::Null; requests[1].count];
        second[0] =
            serde_json::json!({"id":"aaaaaaaaaaa","title":"Newest duplicate","timestamp":600});
        second[1] =
            serde_json::json!({"id":"ddddddddddd","title":"Second dated row","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, &requests).unwrap();
        assert!(page.has_more);
        assert_eq!(
            page.feed_progress
                .iter()
                .map(|entry| entry.count)
                .collect::<Vec<_>>(),
            [12, 12, 1, 1, 0]
        );
        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 requests = FeedCursor::default().plan(3).unwrap();
        let exhausted = feed_results(
            &format!(
                "{}\n{}\n{}",
                short,
                older_short,
                serde_json::json!({"entries":[]})
            ),
            &requests,
        )
        .unwrap();
        assert!(!exhausted.has_more);
        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(),
            ""
        );
    }
}