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 video_urls_preserve_provider_identity_and_reject_invalid_keys() {
    assert_eq!(
        video_url("jNQXAC9IVRw").unwrap(),
        "https://www.youtube.com/watch?v=jNQXAC9IVRw"
    );
    assert_eq!(
        video_url("miruro:opaque_ID-7:3").unwrap(),
        "https://barelystarted.miruro.tv/watch/opaque_ID-7/episode?ep=3"
    );
    assert_eq!(
        video_url("anilist:1:3").unwrap(),
        "https://zokoanime.video/stream/ani/1/3/sub"
    );
    for id in [
        "",
        "--help",
        "jNQXAC9IVRw&list=other",
        "miruro:id:0",
        "miruro:id:03",
        "miruro:../id:3",
        "anilist:01:3",
        "anilist:1:03",
        "anilist:1:0",
        "anilist:1:2147483648",
    ] {
        assert_eq!(
            video_url(id).unwrap_err().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
        ));
    }
    for url in [
        "http://example.test/cover.jpg",
        "file:///tmp/cover.jpg",
        "ftp://example.test/cover.jpg",
        "//example.test/cover.jpg",
        "https://",
        "https://user:secret@example.test/cover.jpg",
        "https://example.test/\ncover.jpg",
    ] {
        assert!(matches!(
            ThumbnailJob::from_url(url),
            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 members_only_rows_are_hidden_without_exhausting_search_or_feed_pages() {
    let members = serde_json::json!({
        "id": "jNQXAC9IVRw", "title": "Members only", "availability": "subscriber_only"
    });
    let mut entries = vec![members; 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);
    let mut cursor = FeedCursor::default();
    let requests = cursor.plan(1).unwrap();
    let feed = feed_results(&output, &requests).unwrap();
    assert!(feed.videos.is_empty());
    assert!(feed.has_more);
    cursor.accept(&feed.feed_progress).unwrap();
    assert_eq!(cursor.plan(1).unwrap()[0].start, PAGE_SIZE + 1);

    for (index, availability) in [
        Some("public"),
        Some("unlisted"),
        None,
        Some("new_availability"),
    ]
    .into_iter()
    .enumerate()
    {
        entries[index] = serde_json::json!({
            "id": format!("{index:011}"),
            "title": "Join my channel membership",
            "availability": availability,
        });
    }
    entries[4] = serde_json::json!({"id": "rQ_J9WH6CGk", "title": "No availability"});
    let output = serde_json::json!({"entries": entries}).to_string();
    for page in [
        search_results(&output).unwrap(),
        feed_results(&output, &cursor.plan(1).unwrap()).unwrap(),
    ] {
        assert_eq!(
            page.videos
                .iter()
                .map(|video| video.id.as_str())
                .collect::<Vec<_>>(),
            [
                "00000000000",
                "00000000001",
                "00000000002",
                "00000000003",
                "rQ_J9WH6CGk"
            ]
        );
        assert!(page.has_more);
    }
}

#[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 channel_playlist_listing_uses_flat_ids_sanitizes_and_deduplicates() {
    // yt-dlp's /playlists tab returns URL rows, not nested playlist contents.
    let playlists = channel_playlists(
        r#"{"_type":"playlist","entries":[
            {"_type":"url","ie_key":"YoutubeTab","id":"PLJ5MFiMfElFA","title":"Rust\u001b"},
            null,42,{},{"id":"PLJ5MFiMfElFA","title":"Duplicate"},
            {"id":"PLabcdefghij&list=other","title":"Unsafe"},
            {"id":"rQ_J9WH6CGk","title":"Video"},
            {"id":"PLabcdefghij","title":null},
            {"id":"PLabcdefghij","title":"\u001b\n"},
            {"id":"OLAK5uy_abcdefghij","title":"Release"}
        ]}"#,
    )
    .unwrap();
    assert_eq!(
        playlists
            .iter()
            .map(|playlist| (playlist.id.as_str(), playlist.title.as_str()))
            .collect::<Vec<_>>(),
        [("PLJ5MFiMfElFA", "Rust"), ("OLAK5uy_abcdefghij", "Release")]
    );
    assert!(channel_playlists(r#"{"entries":[]}"#).unwrap().is_empty());
    for output in ["not JSON", "{}", r#"{"entries":null}"#, r#"{"entries":{}}"#] {
        assert!(channel_playlists(output).is_err());
        assert!(playlist_results(output).is_err());
    }
}

#[test]
fn playlist_ids_are_safe_and_support_music_and_legacy_families() {
    for id in [
        "PLJ5MFiMfElFA",
        "OLAK5uy_abcdefghij-_",
        "UUabcdefghij",
        "RDabcdefghij",
        "RDMM",
        "WL",
        "LL",
        "LM",
    ] {
        assert!(valid_playlist_id(id), "{id}");
        assert!(matches!(
            Job::playlist(id, 0),
            Err(error) if error.kind() == io::ErrorKind::InvalidInput
        ));
    }
    for id in [
        "",
        "--bad",
        "PLshort",
        "PLabcdefghij&x=1",
        "PLabcdefghij#x",
        "PLabcdefghij/",
    ] {
        assert!(!valid_playlist_id(id), "{id}");
        assert!(matches!(
            Job::playlist(id, 1),
            Err(error) if error.kind() == io::ErrorKind::InvalidInput
        ));
    }
    assert!(Job::channel_playlists("https://example.com/@channel").is_err());
}

#[test]
fn playlists_preserve_order_repeats_and_creator_identity_with_raw_pagination() {
    let mut entries = vec![serde_json::Value::Null; PAGE_SIZE];
    entries[0] = serde_json::json!({
        "id":"l-GaaivVJCE","title":"First\u{1b}","timestamp":100
    });
    entries[1] = serde_json::json!({
        "id":"SLV9AJxnb1U","title":"Second","timestamp":300,
        "channel":"Another creator","channel_id":"UC4QobU6STFB0P71PMvOGN5A"
    });
    entries[2] = serde_json::json!({"id":"bad","title":"Invalid"});
    entries[3] = serde_json::json!({"id":"jNQXAC9IVRw","title":null});
    entries[4] = serde_json::json!({
        "id":"jNQXAC9IVRw","title":"Members only","availability":"subscriber_only"
    });
    entries[5] = entries[0].clone();
    entries[6] = serde_json::json!({
        "id":"rQ_J9WH6CGk","title":"Owner upload","channel_id":"UCaYhcUwRBNscFNUKTjgPFiA"
    });
    let output = serde_json::json!({
        "channel_id":"UCaYhcUwRBNscFNUKTjgPFiA","uploader_id":"@RustVideos",
        "channel":"Rust\u{1b}","entries":entries
    });
    let page = playlist_results(&output.to_string()).unwrap();
    assert!(page.has_more);
    assert!(page.feed_progress.is_empty());
    assert_eq!(
        page.videos
            .iter()
            .map(|video| video.id.as_str())
            .collect::<Vec<_>>(),
        ["l-GaaivVJCE", "SLV9AJxnb1U", "l-GaaivVJCE", "rQ_J9WH6CGk"]
    );
    assert_eq!(page.videos[0].title, "First");
    assert_eq!(page.videos[0].channel, None);
    assert_eq!(page.videos[0].channel_id, None);
    assert_eq!(page.videos[0].uploader_id, None);
    assert_eq!(page.videos[3].channel.as_deref(), Some("Rust"));
    assert_eq!(page.videos[3].uploader_id.as_deref(), Some("@RustVideos"));
    assert_eq!(page.videos[1].channel.as_deref(), Some("Another creator"));
    assert_eq!(
        page.videos[1].channel_id.as_deref(),
        Some("UC4QobU6STFB0P71PMvOGN5A")
    );
    assert_eq!(page.videos[1].uploader_id, None);
    assert!(!playlist_results(r#"{"entries":[]}"#).unwrap().has_more);
    assert!(playlist_results(r#"{"channel_id":"invalid","entries":[]}"#).is_err());
    assert!(playlist_results(r#"{"uploader_id":"@bad\u001b","entries":[]}"#).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(),
        ""
    );
}