use std::{
    collections::BTreeSet,
    fs::{self, File, OpenOptions, ReadDir},
    io::{self, Read, Write},
    path::{Path, PathBuf},
    sync::atomic::{AtomicU64, Ordering},
};

use serde::Deserialize;
use toml_edit::{Array, ArrayOfTables, DocumentMut, Table, value};

use crate::youtube::{Video, valid_video_id, validate_video};
use tubular::miruro::{Identity, SourceDocument};

const MAX_STATE_BYTES: u64 = 8 * 1024 * 1024;
const MAX_LINE: usize = 16 * 1024;
const STREAM_BYTES: usize = 16 * 1024;
const MAX_STREAMS: usize = 4096;
const MAX_SOURCE_BYTES: usize = 1024 * 1024;
const MAX_IDENTITY_BYTES: u64 = 7 + 512 + 1 + 10;
static NEXT_NAME: AtomicU64 = AtomicU64::new(0);

fn invalid(message: impl Into<String>) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, message.into())
}

struct StoredFile {
    path: PathBuf,
    original: Option<Vec<u8>>,
}

impl StoredFile {
    fn open(path: PathBuf) -> io::Result<(Self, DocumentMut)> {
        let original = read_bytes(&path, MAX_STATE_BYTES)?;
        let document = match &original {
            Some(bytes) => std::str::from_utf8(bytes)
                .map_err(|e| invalid(format!("{} is not UTF-8: {e}", path.display())))?
                .parse::<DocumentMut>()
                .map_err(|e| {
                    invalid(format!(
                        "Invalid {}: {e}; repair this file before retrying",
                        path.display()
                    ))
                })?,
            None => DocumentMut::new(),
        };
        if original.is_some() && document.get("version").and_then(|v| v.as_integer()) != Some(1) {
            return Err(invalid(format!("{} requires version = 1", path.display())));
        }
        Ok((Self { path, original }, document))
    }

    fn save(&mut self, document: DocumentMut) -> io::Result<()> {
        let contents = document.to_string().into_bytes();
        atomic_write(&self.path, self.original.as_deref(), &contents)?;
        self.original = Some(contents);
        Ok(())
    }
}

#[derive(Clone, Copy)]
pub enum CollectionKind {
    WatchLater,
    Saved,
    Watching,
}

#[derive(Clone, Copy)]
pub enum WatchStatus {
    Watching,
    Watched,
}

struct Collection {
    file: StoredFile,
    videos: Vec<Video>,
    ids: BTreeSet<String>,
    require_channel: bool,
}

impl Collection {
    fn open(path: PathBuf, require_channel: bool) -> io::Result<Self> {
        let (file, document) = StoredFile::open(path)?;
        let (videos, ids) = load_videos(&document, require_channel)
            .map_err(|e| invalid(format!("{}: {e}", file.path.display())))?;
        Ok(Self {
            file,
            videos,
            ids,
            require_channel,
        })
    }

    fn save(&mut self, video: &Video) -> io::Result<()> {
        if self.ids.contains(&video.id) {
            return check_original(&self.file.path, self.file.original.as_deref());
        }
        let mut video = video.clone();
        validate_snapshot(&mut video, self.require_channel)?;
        self.file.save(video_document(
            self.videos.iter().chain(std::iter::once(&video)),
        ))?;
        self.ids.insert(video.id.clone());
        self.videos.push(video);
        Ok(())
    }

    fn remove(&mut self, ids: &[String]) -> io::Result<()> {
        if !ids.iter().any(|id| self.ids.contains(id)) {
            return check_original(&self.file.path, self.file.original.as_deref());
        }
        self.file.save(video_document(
            self.videos.iter().filter(|v| !ids.contains(&v.id)),
        ))?;
        self.videos.retain(|v| !ids.contains(&v.id));
        self.ids.retain(|id| !ids.contains(id));
        Ok(())
    }
}

pub struct State {
    watch_later: Collection,
    saved_videos: Collection,
    watching: Collection,
    watch_state: StoredFile,
    watched: BTreeSet<String>,
    pending: BTreeSet<String>,
    sessions: PathBuf,
    streams: Vec<ProgressStream>,
    discovery: Option<ReadDir>,
    next_stream: usize,
    notify_changed: bool,
}

impl State {
    pub fn open(config_path: &Path) -> io::Result<Self> {
        let parent = config_path
            .parent()
            .filter(|p| !p.as_os_str().is_empty())
            .unwrap_or(Path::new("."));
        let watch_later = Collection::open(parent.join("watch-later.toml"), true)?;
        let saved_videos = Collection::open(parent.join("saved-videos.toml"), true)?;
        let mut watching = Collection::open(parent.join("currently-watching.toml"), false)?;
        let (watch_state, document) = StoredFile::open(parent.join("watch-state.toml"))?;
        let watched = load_watched(&document)
            .map_err(|e| invalid(format!("{}: {e}", watch_state.path.display())))?;
        let completed: Vec<_> = watching.ids.intersection(&watched).cloned().collect();
        if !completed.is_empty() {
            watching.remove(&completed)?;
        }
        Ok(Self {
            watch_later,
            saved_videos,
            watching,
            watch_state,
            watched,
            pending: BTreeSet::new(),
            sessions: parent.join("playback-sessions"),
            streams: Vec::new(),
            discovery: None,
            next_stream: 0,
            notify_changed: false,
        })
    }

    fn collection(&self, kind: CollectionKind) -> &Collection {
        match kind {
            CollectionKind::WatchLater => &self.watch_later,
            CollectionKind::Saved => &self.saved_videos,
            CollectionKind::Watching => &self.watching,
        }
    }

    fn collection_mut(&mut self, kind: CollectionKind) -> &mut Collection {
        match kind {
            CollectionKind::WatchLater => &mut self.watch_later,
            CollectionKind::Saved => &mut self.saved_videos,
            CollectionKind::Watching => &mut self.watching,
        }
    }

    pub fn videos(&self, kind: CollectionKind) -> &[Video] {
        &self.collection(kind).videos
    }
    pub fn contains(&self, kind: CollectionKind, id: &str) -> bool {
        self.collection(kind).ids.contains(id)
    }
    pub fn is_watched(&self, id: &str) -> bool {
        self.watched.contains(id)
    }

    pub fn save(&mut self, kind: CollectionKind, video: &Video) -> io::Result<()> {
        if matches!(kind, CollectionKind::Watching) && self.is_watched(&video.id) {
            return Ok(());
        }
        self.collection_mut(kind).save(video)
    }

    pub fn remove(&mut self, kind: CollectionKind, ids: &[String]) -> io::Result<()> {
        if matches!(kind, CollectionKind::Watching) {
            self.suppress_sessions(ids, false)?;
        }
        self.collection_mut(kind).remove(ids)
    }

    pub fn set_watch_status(&mut self, video: &Video, status: WatchStatus) -> io::Result<()> {
        let mut video = video.clone();
        validate_snapshot(&mut video, false)?;
        check_original(&self.watch_state.path, self.watch_state.original.as_deref())?;
        check_original(
            &self.watching.file.path,
            self.watching.file.original.as_deref(),
        )?;
        let result = (|| {
            self.suppress_sessions(std::slice::from_ref(&video.id), true)?;
            self.pending.remove(&video.id);
            match status {
                WatchStatus::Watching => {
                    self.watching.save(&video)?;
                    self.notify_changed = true;
                    if self.watched.contains(&video.id) {
                        self.watch_state.save(watched_document(
                            self.watched.iter().filter(|id| *id != &video.id),
                        ))?;
                        self.watched.remove(&video.id);
                    }
                }
                WatchStatus::Watched => {
                    if !self.watched.contains(&video.id) {
                        self.watch_state.save(watched_document(
                            self.watched.iter().chain(std::iter::once(&video.id)),
                        ))?;
                        self.watched.insert(video.id.clone());
                        self.notify_changed = true;
                    }
                    self.watching.remove(std::slice::from_ref(&video.id))?;
                    self.notify_changed = true;
                }
            }
            Ok(())
        })();
        result.map_err(|e: io::Error| {
            io::Error::new(
                e.kind(),
                format!(
                    "Could not set watch status; some session/state files may already have been updated: {e}"
                ),
            )
        })
    }

    fn suppress_sessions(&mut self, ids: &[String], status_edited: bool) -> io::Result<()> {
        self.discovery = None;
        loop {
            self.discover()?;
            if self.discovery.is_none() {
                break;
            }
        }
        for stream in &mut self.streams {
            match stream.prepare() {
                Ok(()) => {}
                Err(e) if e.kind() == io::ErrorKind::NotFound => continue,
                Err(e) => return Err(e),
            }
            if stream.id.as_ref().is_some_and(|id| ids.contains(id)) {
                stream.mark_seen(status_edited)?;
            }
        }
        Ok(())
    }

    pub fn progress_file(&mut self, video: &Video) -> io::Result<PathBuf> {
        if self.streams.len() >= MAX_STREAMS {
            return Err(invalid(
                "Playback session limit reached; remove inactive playback-sessions directories and restart",
            ));
        }
        let mut video = video.clone();
        validate_snapshot(&mut video, false)?;
        fs::create_dir_all(self.sessions.parent().unwrap_or(Path::new(".")))?;
        private_directory(&self.sessions, true)?;
        let directory = loop {
            let path = self.sessions.join(format!(
                "session-{}-{}",
                std::process::id(),
                NEXT_NAME.fetch_add(1, Ordering::Relaxed)
            ));
            match create_private_directory(&path) {
                Ok(()) => break path,
                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
                Err(e) => return Err(e),
            }
        };
        let (mut snapshot, _) = StoredFile::open(directory.join("video.toml"))?;
        snapshot.save(video_document(std::iter::once(&video)))?;
        let path = directory.join("progress.jsonl");
        self.streams.push(ProgressStream::new(path.clone()));
        Ok(path)
    }

    pub fn miruro_session(
        &mut self,
        video: &Video,
        source: &SourceDocument,
    ) -> io::Result<(PathBuf, PathBuf)> {
        source.validate()?;
        if Identity::from_key(&video.id)? != source.identity {
            return Err(invalid("Source identity must match the playback episode"));
        }
        let contents = serde_json::to_vec(source).map_err(|e| invalid(e.to_string()))?;
        if contents.len() > MAX_SOURCE_BYTES {
            return Err(invalid("Playback source exceeds the 1 MiB limit"));
        }
        let progress_file = self.progress_file(video)?;
        let source_file = progress_file.with_file_name("source.json");
        atomic_write(&source_file, None, &contents)?;
        register_identity(&progress_file, &video.id)?;
        Ok((source_file, progress_file))
    }

    pub fn poll_progress(&mut self) -> io::Result<bool> {
        let mut error = self.discover().err();
        let count = self.streams.len().min(16);
        for _ in 0..count {
            if self.next_stream >= self.streams.len() {
                self.next_stream = 0;
            }
            let stream = &mut self.streams[self.next_stream];
            self.next_stream += 1;
            if let Err(e) = stream.poll(&mut self.pending, &self.watched)
                && error.is_none()
            {
                error = Some(e);
            }
        }
        self.pending.retain(|id| !self.watched.contains(id));
        if !self.pending.is_empty() {
            self.watch_state.save(watched_document(
                self.watched.iter().chain(self.pending.iter()),
            ))?;
            self.watched.append(&mut self.pending);
            self.notify_changed = true;
        }
        let completed: Vec<_> = self
            .watching
            .ids
            .intersection(&self.watched)
            .cloned()
            .collect();
        if !completed.is_empty() {
            self.watching.remove(&completed)?;
            self.notify_changed = true;
        }
        for stream in &mut self.streams {
            if !stream.started || stream.seen {
                continue;
            }
            let id = stream.id.as_ref().unwrap();
            if !self.watched.contains(id) {
                let video = stream.video.as_ref().or_else(|| {
                    self.watch_later
                        .videos
                        .iter()
                        .chain(self.saved_videos.videos.iter())
                        .find(|video| &video.id == id)
                });
                if let Some(video) = video {
                    let added = !self.watching.ids.contains(id);
                    self.watching.save(video)?;
                    self.notify_changed |= added;
                }
                stream.mark_seen(false)?;
            }
            stream.started = false;
        }
        // Preserve notification after reporting a bad record; siblings were still processed.
        if let Some(error) = error {
            return Err(error);
        }
        Ok(std::mem::take(&mut self.notify_changed))
    }

    fn discover(&mut self) -> io::Result<()> {
        match private_directory(&self.sessions, false) {
            Ok(()) => {}
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
            Err(e) => return Err(e),
        }
        if self.discovery.is_none() {
            self.discovery = Some(fs::read_dir(&self.sessions)?);
        }
        for _ in 0..32 {
            let Some(entry) = self.discovery.as_mut().unwrap().next() else {
                self.discovery = None;
                break;
            };
            let entry = entry?;
            if !entry.file_name().to_string_lossy().starts_with("session-") {
                continue;
            }
            let path = entry.path().join("progress.jsonl");
            if self.streams.iter().any(|stream| stream.path == path) {
                continue;
            }
            if private_directory(&entry.path(), false).is_err() {
                continue;
            }
            // Failed launchers leave empty private directories; only bound identities are discoverable.
            if !matches!(fs::symlink_metadata(entry.path().join("video-id")), Ok(m) if m.is_file())
            {
                continue;
            }
            if self.streams.len() >= MAX_STREAMS {
                return Err(invalid(
                    "Playback session limit reached; remove inactive playback-sessions directories and restart",
                ));
            }
            self.streams.push(ProgressStream::new(path));
        }
        // ponytail: retain reports for live GUI writers, capped at 4096 monitored sessions;
        // add explicit inactive-session archival if long-lived installations hit this ceiling.
        Ok(())
    }
}

fn validate_snapshot(video: &mut Video, require_channel: bool) -> io::Result<()> {
    let date = video.publication_date.clone();
    if let Some(date) = &date {
        let raw = date.strip_prefix('~').unwrap_or(date);
        let compact = raw.replace('-', "");
        if !(raw.len() == 10 && raw.as_bytes()[4] == b'-' && raw.as_bytes()[7] == b'-')
            || compact.len() != 8
            || !compact.bytes().all(|b| b.is_ascii_digit())
            || time::Date::from_calendar_date(
                compact[..4]
                    .parse()
                    .map_err(|_| invalid("Invalid saved publication date"))?,
                compact[4..6]
                    .parse::<u8>()
                    .ok()
                    .and_then(|m| m.try_into().ok())
                    .ok_or_else(|| invalid("Invalid saved publication month"))?,
                compact[6..]
                    .parse()
                    .map_err(|_| invalid("Invalid saved publication day"))?,
            )
            .is_err()
        {
            return Err(invalid(
                "Saved publication date must be a valid YYYY-MM-DD or ~YYYY-MM-DD",
            ));
        }
    }
    if video.id.starts_with("miruro:") {
        Identity::from_key(&video.id)?;
        if video.title.trim().is_empty() || video.title.chars().any(char::is_control) {
            return Err(invalid(
                "Miruro episode title must be nonempty and contain no controls",
            ));
        }
        if video.channel.as_deref() != Some("Miruro")
            || video.channel_id.is_some()
            || video.uploader_id.is_some()
        {
            return Err(invalid(
                "Miruro episodes must not contain YouTube channel identities",
            ));
        }
        return Ok(());
    }
    validate_video(video).map_err(invalid)?;
    video.publication_date = date;
    if let Some(handle) = &video.uploader_id
        && (!handle.starts_with('@') || crate::config::normalize_channel(handle).is_err())
    {
        return Err(invalid("Saved uploader_id must be a valid YouTube @handle"));
    }
    if require_channel && video.channel_id.is_none() && video.uploader_id.is_none() {
        return Err(invalid(
            "Video has no channel identity; refresh its metadata before saving",
        ));
    }
    Ok(())
}

fn load_videos(
    document: &DocumentMut,
    require_channel: bool,
) -> io::Result<(Vec<Video>, BTreeSet<String>)> {
    if document
        .iter()
        .any(|(key, _)| !matches!(key, "version" | "videos"))
    {
        return Err(invalid("Unknown watch-later field"));
    }
    let Some(item) = document.get("videos") else {
        return Ok((Vec::new(), BTreeSet::new()));
    };
    if item.as_array().is_some_and(|a| a.is_empty()) {
        return Ok((Vec::new(), BTreeSet::new()));
    }
    let tables = item
        .as_array_of_tables()
        .ok_or_else(|| invalid("videos must be an array of tables"))?;
    let mut videos = Vec::new();
    let mut ids = BTreeSet::new();
    for table in tables {
        let mut fields = serde_json::Map::new();
        for (key, item) in table.iter() {
            let field = match key {
                "id" | "title" | "channel" | "channel_id" | "uploader_id" | "upload_date" => {
                    serde_json::Value::String(
                        item.as_str()
                            .ok_or_else(|| invalid(format!("Video {key} must be a string")))?
                            .into(),
                    )
                }
                "timestamp" => serde_json::Value::from(
                    item.as_integer()
                        .ok_or_else(|| invalid("Video timestamp must be an integer"))?,
                ),
                _ => return Err(invalid(format!("Unknown saved video field {key:?}"))),
            };
            fields.insert(key.into(), field);
        }
        let mut video: Video = serde_json::from_value(serde_json::Value::Object(fields))
            .map_err(|e| invalid(format!("Invalid saved video: {e}")))?;
        validate_snapshot(&mut video, require_channel)?;
        if ids.insert(video.id.clone()) {
            videos.push(video);
        }
    }
    Ok((videos, ids))
}

fn load_watched(document: &DocumentMut) -> io::Result<BTreeSet<String>> {
    if document
        .iter()
        .any(|(key, _)| !matches!(key, "version" | "watched"))
    {
        return Err(invalid("Unknown watch-state field"));
    }
    let Some(item) = document.get("watched") else {
        return Ok(BTreeSet::new());
    };
    item.as_array()
        .ok_or_else(|| invalid("watched must be an array of playback identities"))?
        .iter()
        .map(|v| {
            let id = v
                .as_str()
                .filter(|id| valid_identity(id))
                .ok_or_else(|| invalid("Invalid watched playback identity"))?;
            Ok(id.to_owned())
        })
        .collect()
}

fn video_document<'a>(videos: impl Iterator<Item = &'a Video>) -> DocumentMut {
    let mut document = DocumentMut::new();
    document["version"] = value(1);
    let mut tables = ArrayOfTables::new();
    for video in videos {
        let mut table = Table::new();
        table["id"] = value(video.id.as_str());
        table["title"] = value(video.title.as_str());
        for (key, field) in [
            ("channel", &video.channel),
            ("channel_id", &video.channel_id),
            ("uploader_id", &video.uploader_id),
            ("upload_date", &video.publication_date),
        ] {
            if let Some(field) = field {
                table[key] = value(field.as_str());
            }
        }
        if let Some(timestamp) = video.timestamp {
            table["timestamp"] = value(timestamp);
        }
        tables.push(table);
    }
    document["videos"] = if tables.is_empty() {
        value(Array::new())
    } else {
        toml_edit::Item::ArrayOfTables(tables)
    };
    document
}

fn watched_document<'a>(watched: impl Iterator<Item = &'a String>) -> DocumentMut {
    let mut document = DocumentMut::new();
    document["version"] = value(1);
    let mut ids = Array::new();
    for id in watched {
        ids.push(id.as_str());
    }
    document["watched"] = value(ids);
    document
}

fn regular_file(path: &Path, private: bool) -> io::Result<File> {
    let mut options = OpenOptions::new();
    options.read(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
    }
    let file = options.open(path)?;
    let metadata = file.metadata()?;
    if !metadata.is_file() {
        return Err(invalid(format!(
            "{} must be a regular file",
            path.display()
        )));
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        // SAFETY: geteuid has no preconditions.
        if metadata.uid() != unsafe { libc::geteuid() } || (private && metadata.mode() & 0o077 != 0)
        {
            return Err(invalid(format!(
                "{} must be owned by you with owner-only permissions",
                path.display()
            )));
        }
    }
    Ok(file)
}

fn read_bytes(path: &Path, limit: u64) -> io::Result<Option<Vec<u8>>> {
    let file = match regular_file(path, false) {
        Ok(file) => file,
        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e),
    };
    let mut bytes = Vec::new();
    file.take(limit + 1).read_to_end(&mut bytes)?;
    if bytes.len() as u64 > limit {
        return Err(invalid(format!(
            "{} exceeds the {} byte limit",
            path.display(),
            limit
        )));
    }
    Ok(Some(bytes))
}

fn check_original(path: &Path, original: Option<&[u8]>) -> io::Result<()> {
    if read_bytes(path, MAX_STATE_BYTES)?.as_deref() != original {
        return Err(io::Error::other(format!(
            "{} changed on disk; restart Tubular before retrying (no changes written)",
            path.display()
        )));
    }
    Ok(())
}

fn atomic_write(path: &Path, original: Option<&[u8]>, contents: &[u8]) -> io::Result<()> {
    if contents.len() as u64 > MAX_STATE_BYTES {
        return Err(invalid("Watch state exceeds the 8 MiB limit"));
    }
    let parent = path.parent().unwrap_or(Path::new("."));
    fs::create_dir_all(parent)?;
    let (temporary, mut file) = loop {
        let path = parent.join(format!(
            ".watch-{}-{}.tmp",
            std::process::id(),
            NEXT_NAME.fetch_add(1, Ordering::Relaxed)
        ));
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        match options.open(&path) {
            Ok(file) => break (path, file),
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
            Err(e) => return Err(e),
        }
    };
    let result = (|| {
        file.write_all(contents)?;
        file.sync_all()?;
        check_original(path, original)?;
        if original.is_none() {
            fs::hard_link(&temporary, path)?;
        } else {
            fs::rename(&temporary, path)?;
        }
        Ok(())
    })();
    drop(file);
    let _ = fs::remove_file(temporary);
    result
}

fn create_private_directory(path: &Path) -> io::Result<()> {
    let mut builder = fs::DirBuilder::new();
    #[cfg(unix)]
    {
        use std::os::unix::fs::DirBuilderExt;
        builder.mode(0o700);
    }
    builder.create(path)
}

fn private_directory(path: &Path, create: bool) -> io::Result<()> {
    if create {
        match create_private_directory(path) {
            Ok(()) => {}
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
            Err(e) => return Err(e),
        }
    }
    let metadata = fs::symlink_metadata(path)?;
    if !metadata.is_dir() {
        return Err(invalid(format!(
            "{} must be a plain private directory, not a symlink",
            path.display()
        )));
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        // SAFETY: geteuid has no preconditions.
        if metadata.uid() != unsafe { libc::geteuid() } || metadata.mode() & 0o077 != 0 {
            return Err(invalid(format!(
                "{} must be owned by you with permissions 0700",
                path.display()
            )));
        }
    }
    Ok(())
}

fn valid_identity(id: &str) -> bool {
    valid_video_id(id) || Identity::from_key(id).is_ok()
}

pub(crate) fn register_identity(progress_file: &Path, id: &str) -> io::Result<()> {
    if !valid_identity(id)
        || progress_file
            .file_name()
            .is_none_or(|name| name != "progress.jsonl")
    {
        return Err(invalid(
            "Invalid playback reporting identity or destination",
        ));
    }
    let directory = progress_file
        .parent()
        .ok_or_else(|| invalid("Missing private session directory"))?;
    private_directory(directory, false)?;
    private_directory(
        directory
            .parent()
            .ok_or_else(|| invalid("Missing playback-sessions directory"))?,
        false,
    )?;
    match fs::symlink_metadata(progress_file) {
        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
        Ok(_) => {
            return Err(invalid(
                "Playback report destination must not already exist",
            ));
        }
        Err(e) => return Err(e),
    }
    atomic_write(&directory.join("video-id"), None, id.as_bytes())
}

#[derive(Deserialize)]
#[serde(untagged)]
enum ProgressRecord {
    YouTube(YouTubeProgress),
    Miruro(MiruroProgress),
}

#[derive(Deserialize)]
struct YouTubeProgress {
    version: u32,
    event: String,
    media_url: String,
    position_seconds: f64,
    duration_seconds: Option<f64>,
    completion_eligible: bool,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct MiruroProgress {
    version: u32,
    event: String,
    media_identity: Identity,
    position_seconds: f64,
    duration_seconds: Option<f64>,
    completion_eligible: bool,
}

struct ProgressStream {
    path: PathBuf,
    id: Option<String>,
    video: Option<Video>,
    seen_file: Option<StoredFile>,
    seen: bool,
    status_edited: bool,
    started: bool,
    file: Option<File>,
    partial: Vec<u8>,
    discarding: bool,
    failed: bool,
    line: u64,
}

impl ProgressStream {
    fn new(path: PathBuf) -> Self {
        Self {
            path,
            id: None,
            video: None,
            seen_file: None,
            seen: false,
            status_edited: false,
            started: false,
            file: None,
            partial: Vec::new(),
            discarding: false,
            failed: false,
            line: 0,
        }
    }

    fn prepare(&mut self) -> io::Result<()> {
        if self.id.is_some() {
            return Ok(());
        }
        private_directory(self.path.parent().unwrap(), false)?;
        let marker = self.path.with_file_name("video-id");
        let mut bytes = Vec::new();
        regular_file(&marker, true)?
            .take(MAX_IDENTITY_BYTES + 1)
            .read_to_end(&mut bytes)?;
        let id = std::str::from_utf8(&bytes)
            .ok()
            .filter(|id| id.len() as u64 <= MAX_IDENTITY_BYTES && valid_identity(id))
            .ok_or_else(|| invalid("Invalid video-id session marker"))?;
        let snapshot = self.path.with_file_name("video.toml");
        let video = match regular_file(&snapshot, true) {
            Ok(_) => {
                let mut collection = Collection::open(snapshot, false)?;
                if collection.videos.len() != 1 || collection.videos[0].id != id {
                    return Err(invalid(
                        "Session video metadata must match its video-id marker",
                    ));
                }
                collection.videos.pop()
            }
            Err(e) if e.kind() == io::ErrorKind::NotFound && valid_video_id(id) => None,
            Err(e) if e.kind() == io::ErrorKind::NotFound => {
                return Err(invalid(
                    "Miruro sessions require their bound video metadata",
                ));
            }
            Err(e) => return Err(e),
        };
        let seen_path = self.path.with_file_name("watching-seen.toml");
        match regular_file(&seen_path, true) {
            Ok(_) => {}
            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
            Err(e) => return Err(e),
        }
        let (seen_file, document) = StoredFile::open(seen_path)?;
        let seen = seen_file.original.is_some();
        if seen
            && (document
                .iter()
                .any(|(key, _)| !matches!(key, "version" | "id" | "status_edited"))
                || document.get("id").and_then(|item| item.as_str()) != Some(id))
        {
            return Err(invalid(
                "Session watching admission must match its video-id marker",
            ));
        }
        let status_edited = match document.get("status_edited") {
            Some(item) => item
                .as_bool()
                .ok_or_else(|| invalid("Session status_edited must be a boolean"))?,
            None => false,
        };
        self.id = Some(id.into());
        self.video = video;
        self.seen_file = Some(seen_file);
        self.seen = seen;
        self.status_edited = status_edited;
        Ok(())
    }

    fn mark_seen(&mut self, status_edited: bool) -> io::Result<()> {
        if self.seen && (!status_edited || self.status_edited) {
            let file = self.seen_file.as_ref().unwrap();
            return check_original(&file.path, file.original.as_deref());
        }
        private_directory(self.path.parent().unwrap(), false)?;
        let mut document = DocumentMut::new();
        document["version"] = value(1);
        document["id"] = value(self.id.as_deref().unwrap());
        if status_edited || self.status_edited {
            document["status_edited"] = value(true);
        }
        self.seen_file.as_mut().unwrap().save(document)?;
        self.seen = true;
        self.status_edited |= status_edited;
        self.started = false;
        Ok(())
    }

    fn poll(
        &mut self,
        pending: &mut BTreeSet<String>,
        watched: &BTreeSet<String>,
    ) -> io::Result<()> {
        if self.failed {
            return Ok(());
        }
        match self.read(pending, watched) {
            Ok(error) => match error {
                Some(e) => Err(e),
                None => Ok(()),
            },
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
            Err(e) => {
                self.failed = true;
                Err(invalid(format!("Cannot read {}: {e}", self.path.display())))
            }
        }
    }

    fn read(
        &mut self,
        pending: &mut BTreeSet<String>,
        watched: &BTreeSet<String>,
    ) -> io::Result<Option<io::Error>> {
        if self.file.is_none() {
            self.prepare()?;
            self.file = Some(regular_file(&self.path, true)?);
        }
        let mut buffer = [0; STREAM_BYTES];
        let size = self.file.as_mut().unwrap().read(&mut buffer)?;
        let mut error = None;
        for &byte in &buffer[..size] {
            if byte == b'\n' {
                self.line += 1;
                if !self.discarding
                    && let Err(e) = self.record(pending, watched)
                    && error.is_none()
                {
                    error = Some(e);
                }
                self.partial.clear();
                self.discarding = false;
            } else if !self.discarding {
                if self.partial.len() == MAX_LINE {
                    self.partial.clear();
                    self.discarding = true;
                    if error.is_none() {
                        error = Some(self.record_error("line exceeds 16 KiB"));
                    }
                } else {
                    self.partial.push(byte);
                }
            }
        }
        Ok(error)
    }

    fn record_error(&self, detail: impl std::fmt::Display) -> io::Error {
        invalid(format!(
            "Invalid playback report {} line {}: {detail}; inspect this record (other sessions continue)",
            self.path.display(),
            self.line.max(1)
        ))
    }

    fn record(
        &mut self,
        pending: &mut BTreeSet<String>,
        watched: &BTreeSet<String>,
    ) -> io::Result<()> {
        let record: ProgressRecord =
            serde_json::from_slice(&self.partial).map_err(|e| self.record_error(e))?;
        let miruro_key;
        let (id, event, position, duration, eligible) = match &record {
            ProgressRecord::YouTube(record) if record.version == 1 => {
                let id = record
                    .media_url
                    .strip_prefix("https://www.youtube.com/watch?v=")
                    .filter(|id| valid_video_id(id))
                    .ok_or_else(|| self.record_error("expected canonical YouTube video URL"))?;
                (
                    id,
                    record.event.as_str(),
                    record.position_seconds,
                    record.duration_seconds,
                    record.completion_eligible,
                )
            }
            ProgressRecord::Miruro(record) if record.version == 2 => {
                record
                    .media_identity
                    .validate()
                    .map_err(|e| self.record_error(e))?;
                miruro_key = record.media_identity.key();
                (
                    miruro_key.as_str(),
                    record.event.as_str(),
                    record.position_seconds,
                    record.duration_seconds,
                    record.completion_eligible,
                )
            }
            _ => {
                return Err(self.record_error(
                    "expected version 1 with media_url or version 2 with media_identity",
                ));
            }
        };
        if !matches!(event, "progress" | "ended" | "closed") {
            return Err(self.record_error("expected progress, ended, or closed event"));
        }
        if !position.is_finite()
            || position < 0.0
            || duration.is_some_and(|d| !d.is_finite() || d <= 0.0)
        {
            return Err(self.record_error(
                "position must be finite/nonnegative and duration positive/finite or null",
            ));
        }
        if self.id.as_deref() != Some(id) || self.status_edited {
            return Ok(());
        }
        if position > 0.0 && !self.seen && !watched.contains(id) {
            self.started = true;
        }
        if eligible
            && duration.is_some_and(|d| position / d >= 0.95)
            && !watched.contains(id)
            && !pending.contains(id)
        {
            pending.insert(id.into());
        }
        Ok(())
    }
}

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

    struct TestDirectory(PathBuf);
    impl TestDirectory {
        fn new() -> Self {
            loop {
                let path = std::env::temp_dir().join(format!(
                    "tubular-watch-test-{}-{}",
                    std::process::id(),
                    NEXT_NAME.fetch_add(1, Ordering::Relaxed)
                ));
                match create_private_directory(&path) {
                    Ok(()) => return Self(path),
                    Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
                    Err(e) => panic!("{e}"),
                }
            }
        }
        fn config(&self) -> PathBuf {
            self.0.join("config.toml")
        }
    }
    impl Drop for TestDirectory {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.0);
        }
    }

    fn video() -> Video {
        Video {
            id: "jNQXAC9IVRw".into(),
            title: "Me at the zoo".into(),
            channel: Some("jawed".into()),
            channel_id: Some("UC4QobU6STFB0P71PMvOGN5A".into()),
            uploader_id: Some("@jawed".into()),
            timestamp: Some(1_114_300_000),
            publication_date: Some("2005-04-23".into()),
        }
    }

    fn append(path: &Path, bytes: &[u8]) {
        let mut options = OpenOptions::new();
        options.write(true).append(true).create(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        options.open(path).unwrap().write_all(bytes).unwrap();
    }

    fn record(id: &str, event: &str, eligible: bool, position: f64) -> String {
        format!(
            "{{\"version\":1,\"event\":\"{event}\",\"media_url\":\"https://www.youtube.com/watch?v={id}\",\"position_seconds\":{position},\"duration_seconds\":100,\"completion_eligible\":{eligible}}}\n"
        )
    }

    fn miruro_video(episode: u32) -> Video {
        Video {
            id: Identity::new("opaque_Show-1".into(), episode)
                .unwrap()
                .key(),
            title: format!("Example show · Episode {episode}"),
            channel: Some("Miruro".into()),
            channel_id: None,
            uploader_id: None,
            timestamp: Some(1_114_300_000),
            publication_date: Some("2005-04-23".into()),
        }
    }

    fn miruro_source(video: &Video) -> SourceDocument {
        SourceDocument {
            version: 1,
            url: "https://cdn.example/video.m3u8?token=private-token".into(),
            headers: [("Referer".into(), "https://player.example/".into())].into(),
            identity: Identity::from_key(&video.id).unwrap(),
            title: Some(video.title.clone()),
            subtitles: Vec::new(),
        }
    }

    fn miruro_record(
        video: &Video,
        event: &str,
        eligible: bool,
        position: f64,
        duration: Option<f64>,
    ) -> String {
        serde_json::json!({
            "version": 2,
            "event": event,
            "media_identity": Identity::from_key(&video.id).unwrap(),
            "position_seconds": position,
            "duration_seconds": duration,
            "completion_eligible": eligible,
        })
        .to_string()
            + "\n"
    }

    #[test]
    fn manual_status_transitions_preserve_both_providers_and_other_collections() {
        for video in [video(), miruro_video(1)] {
            let directory = TestDirectory::new();
            let mut state = State::open(&directory.config()).unwrap();
            let other = miruro_video(2);
            state
                .set_watch_status(&other, WatchStatus::Watched)
                .unwrap();
            for kind in [CollectionKind::Saved, CollectionKind::WatchLater] {
                state.save(kind, &video).unwrap();
            }
            let saved = fs::read(directory.0.join("saved-videos.toml")).unwrap();
            let later = fs::read(directory.0.join("watch-later.toml")).unwrap();
            for status in [
                WatchStatus::Watching,
                WatchStatus::Watched,
                WatchStatus::Watching,
            ] {
                state.set_watch_status(&video, status).unwrap();
                let watching = fs::read(directory.0.join("currently-watching.toml")).unwrap();
                let watched =
                    read_bytes(&directory.0.join("watch-state.toml"), MAX_STATE_BYTES).unwrap();
                state.set_watch_status(&video, status).unwrap();
                assert_eq!(
                    fs::read(directory.0.join("currently-watching.toml")).unwrap(),
                    watching
                );
                assert_eq!(
                    read_bytes(&directory.0.join("watch-state.toml"), MAX_STATE_BYTES).unwrap(),
                    watched
                );
                state = State::open(&directory.config()).unwrap();
                assert!(state.is_watched(&other.id));
                assert_eq!(
                    state.is_watched(&video.id),
                    matches!(status, WatchStatus::Watched)
                );
                assert_eq!(
                    state.contains(CollectionKind::Watching, &video.id),
                    matches!(status, WatchStatus::Watching)
                );
                if matches!(status, WatchStatus::Watching) {
                    assert_eq!(
                        serde_json::to_value(&state.videos(CollectionKind::Watching)[0]).unwrap(),
                        serde_json::to_value(&video).unwrap()
                    );
                }
                assert_eq!(
                    fs::read(directory.0.join("saved-videos.toml")).unwrap(),
                    saved
                );
                assert_eq!(
                    fs::read(directory.0.join("watch-later.toml")).unwrap(),
                    later
                );
            }
        }
    }

    #[test]
    fn manual_watching_ignores_old_completion_replays_but_accepts_new_sessions() {
        for video in [video(), miruro_video(1)] {
            let directory = TestDirectory::new();
            let mut state = State::open(&directory.config()).unwrap();
            let completion = if video.id.starts_with("miruro:") {
                miruro_record(&video, "ended", true, 95.0, Some(100.0))
            } else {
                record(&video.id, "ended", true, 95.0)
            };
            let first = state.progress_file(&video).unwrap();
            register_identity(&first, &video.id).unwrap();
            append(&first, completion.as_bytes());
            assert!(state.poll_progress().unwrap());
            assert!(state.is_watched(&video.id));
            let second = state.progress_file(&video).unwrap();
            register_identity(&second, &video.id).unwrap();
            // Neither the reports nor the second session have been read by this instance.
            drop(state);
            let mut state = State::open(&directory.config()).unwrap();
            state
                .set_watch_status(&video, WatchStatus::Watching)
                .unwrap();
            for report in [&first, &second] {
                append(report, completion.as_bytes());
            }
            drop(state);
            let mut state = State::open(&directory.config()).unwrap();
            assert!(!state.poll_progress().unwrap());
            assert!(!state.is_watched(&video.id));
            assert!(state.contains(CollectionKind::Watching, &video.id));
            let new_session = state.progress_file(&video).unwrap();
            register_identity(&new_session, &video.id).unwrap();
            append(&new_session, completion.as_bytes());
            assert!(state.poll_progress().unwrap());
            assert!(state.is_watched(&video.id));
            assert!(!state.contains(CollectionKind::Watching, &video.id));
        }
    }

    #[test]
    fn manual_status_rejects_invalid_videos_and_external_edits_before_writing() {
        for video in [video(), miruro_video(1)] {
            for status in [WatchStatus::Watching, WatchStatus::Watched] {
                let directory = TestDirectory::new();
                let mut state = State::open(&directory.config()).unwrap();
                let mut invalid_video = video.clone();
                invalid_video.id = "--invalid".into();
                assert!(state.set_watch_status(&invalid_video, status).is_err());
                assert!(!directory.0.join("watch-state.toml").exists());
                assert!(!directory.0.join("currently-watching.toml").exists());
                state
                    .set_watch_status(&video, WatchStatus::Watching)
                    .unwrap();
                state
                    .set_watch_status(&video, WatchStatus::Watched)
                    .unwrap();
                let report = state.progress_file(&video).unwrap();
                register_identity(&report, &video.id).unwrap();
                for name in ["watch-state.toml", "currently-watching.toml"] {
                    let path = directory.0.join(name);
                    let original = fs::read(&path).unwrap();
                    append(&path, b"\n# external edit\n");
                    let changed = fs::read(&path).unwrap();
                    assert!(state.set_watch_status(&video, status).is_err());
                    assert_eq!(fs::read(&path).unwrap(), changed);
                    assert!(state.is_watched(&video.id));
                    assert!(!state.contains(CollectionKind::Watching, &video.id));
                    assert!(!report.with_file_name("watching-seen.toml").exists());
                    fs::write(path, original).unwrap();
                }
            }
        }
    }

    #[test]
    fn manual_watching_clears_consumed_pending_completion_and_checks_session_edits() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        append(&report, record(&video.id, "ended", true, 95.0).as_bytes());
        let watched = directory.0.join("watch-state.toml");
        fs::write(&watched, "external edit").unwrap();
        assert!(state.poll_progress().is_err());
        assert!(state.pending.contains(&video.id));
        fs::remove_file(watched).unwrap();
        let marker = report.with_file_name("watching-seen.toml");
        fs::write(&marker, "external edit").unwrap();
        assert!(
            state
                .set_watch_status(&video, WatchStatus::Watching)
                .is_err()
        );
        assert!(state.videos(CollectionKind::Watching).is_empty());
        assert_eq!(fs::read_to_string(&marker).unwrap(), "external edit");
        fs::remove_file(&marker).unwrap();
        state
            .set_watch_status(&video, WatchStatus::Watching)
            .unwrap();
        assert!(!state.pending.contains(&video.id));
        state.poll_progress().unwrap();
        assert!(!state.is_watched(&video.id));
        assert!(state.contains(CollectionKind::Watching, &video.id));
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(!state.poll_progress().unwrap());
        assert!(!state.is_watched(&video.id));
        assert!(state.contains(CollectionKind::Watching, &video.id));
    }

    #[test]
    fn collections_are_independent_and_preserve_metadata_and_watched_on_restart() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
            state.save(kind, &video).unwrap();
            state.save(kind, &video).unwrap();
            assert_eq!(state.videos(kind).len(), 1);
            assert!(state.contains(kind, &video.id));
        }
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        append(
            &report,
            record(&video.id, "progress", true, 95.0).as_bytes(),
        );
        assert!(state.poll_progress().unwrap());
        assert!(!state.poll_progress().unwrap());
        let mut restarted = State::open(&directory.config()).unwrap();
        for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
            let stored = &restarted.videos(kind)[0];
            assert_eq!(stored.id, video.id);
            assert_eq!(stored.title, video.title);
            assert_eq!(stored.channel, video.channel);
            assert_eq!(stored.channel_id, video.channel_id);
            assert_eq!(stored.uploader_id, video.uploader_id);
            assert_eq!(stored.timestamp, video.timestamp);
            assert_eq!(stored.publication_date, video.publication_date);
        }
        assert!(restarted.is_watched(&video.id));
        restarted
            .remove(CollectionKind::WatchLater, std::slice::from_ref(&video.id))
            .unwrap();
        restarted
            .remove(CollectionKind::WatchLater, std::slice::from_ref(&video.id))
            .unwrap();
        let mut restarted = State::open(&directory.config()).unwrap();
        assert!(restarted.videos(CollectionKind::WatchLater).is_empty());
        assert!(restarted.contains(CollectionKind::Saved, &video.id));
        assert!(restarted.is_watched(&video.id));
        restarted.save(CollectionKind::WatchLater, &video).unwrap();
        restarted
            .remove(CollectionKind::Saved, std::slice::from_ref(&video.id))
            .unwrap();
        let mut restarted = State::open(&directory.config()).unwrap();
        assert!(restarted.videos(CollectionKind::Saved).is_empty());
        assert!(restarted.contains(CollectionKind::WatchLater, &video.id));
        assert!(restarted.is_watched(&video.id));
        assert!(!restarted.poll_progress().unwrap());
    }

    #[test]
    fn malformed_and_externally_changed_state_is_never_overwritten() {
        let directory = TestDirectory::new();
        let path = directory.0.join("watch-later.toml");
        let mut state = State::open(&directory.config()).unwrap();
        fs::write(&path, "not = [valid").unwrap();
        assert!(state.save(CollectionKind::WatchLater, &video()).is_err());
        assert!(state.videos(CollectionKind::WatchLater).is_empty());
        assert_eq!(fs::read_to_string(&path).unwrap(), "not = [valid");
        assert!(State::open(&directory.config()).is_err());
        fs::remove_file(&path).unwrap();
        let saved_path = directory.0.join("saved-videos.toml");
        fs::write(&saved_path, "not = [valid").unwrap();
        assert!(state.save(CollectionKind::Saved, &video()).is_err());
        assert!(state.videos(CollectionKind::Saved).is_empty());
        assert_eq!(fs::read_to_string(&saved_path).unwrap(), "not = [valid");
        assert!(State::open(&directory.config()).is_err());
        fs::remove_file(&saved_path).unwrap();
        fs::write(&saved_path, "version = 1\n[[videos]]\nid = '--unsafe'\n").unwrap();
        assert!(State::open(&directory.config()).is_err());
        assert_eq!(
            fs::read_to_string(&saved_path).unwrap(),
            "version = 1\n[[videos]]\nid = '--unsafe'\n"
        );
        fs::remove_file(&saved_path).unwrap();
        fs::write(
            directory.0.join("watch-state.toml"),
            "version = 1\nwatched = [\"--unsafe\"]\n",
        )
        .unwrap();
        assert!(State::open(&directory.config()).is_err());
        assert_eq!(
            fs::read_to_string(directory.0.join("watch-state.toml")).unwrap(),
            "version = 1\nwatched = [\"--unsafe\"]\n"
        );
    }

    #[test]
    fn watched_write_failure_keeps_memory_unchanged_and_retries_consumed_reports() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let report = state.progress_file(&video()).unwrap();
        register_identity(&report, &video().id).unwrap();
        append(
            &report,
            record(&video().id, "progress", true, 95.0).as_bytes(),
        );
        let path = directory.0.join("watch-state.toml");
        fs::write(&path, "malformed state").unwrap();
        assert!(state.poll_progress().is_err());
        assert!(!state.is_watched(&video().id));
        assert!(state.videos(CollectionKind::Watching).is_empty());
        assert_eq!(fs::read_to_string(&path).unwrap(), "malformed state");
        fs::remove_file(&path).unwrap();
        assert!(state.poll_progress().unwrap());
        assert!(state.is_watched(&video().id));
        assert!(!state.poll_progress().unwrap());
    }

    #[test]
    fn partial_reports_survive_restart_and_bad_siblings_do_not_block_completion() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let report = state.progress_file(&video()).unwrap();
        register_identity(&report, &video().id).unwrap();
        let line = record(&video().id, "ended", true, 95.0);
        append(&report, &line.as_bytes()[..line.len() - 1]);
        assert!(!state.poll_progress().unwrap());
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(!state.poll_progress().unwrap());
        let mut other = video();
        other.id = "rQ_J9WH6CGk".into();
        let bad = state.progress_file(&other).unwrap();
        register_identity(&bad, &other.id).unwrap();
        append(&bad, b"not json\n");
        append(&report, b"\n");
        assert!(state.poll_progress().is_err());
        assert!(state.is_watched(&video().id));
        assert!(state.poll_progress().unwrap());
        assert!(!state.poll_progress().unwrap());
    }

    #[test]
    fn wrong_identity_ineligible_positions_and_untrusted_reports_do_not_mark_watched() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let report = state.progress_file(&video()).unwrap();
        register_identity(&report, &video().id).unwrap();
        for line in [
            record("rQ_J9WH6CGk", "progress", true, 100.0),
            record(&video().id, "progress", false, 100.0),
            record(&video().id, "closed", true, 94.9),
        ] {
            append(&report, line.as_bytes());
        }
        assert!(state.poll_progress().unwrap());
        assert!(state.contains(CollectionKind::Watching, &video().id));
        assert!(!state.is_watched(&video().id));
        assert!(!state.is_watched("rQ_J9WH6CGk"));
        for line in [
            record(&video().id, "unknown", true, 100.0),
            record(&video().id, "progress", true, -1.0),
            record(&video().id, "progress", true, 100.0).replace("\"version\":1", "\"version\":2"),
            record(&video().id, "progress", true, 100.0)
                .replace("\"duration_seconds\":100", "\"duration_seconds\":0"),
        ] {
            append(&report, line.as_bytes());
            assert!(state.poll_progress().is_err());
            assert!(!state.is_watched(&video().id));
            assert!(!state.poll_progress().unwrap());
        }
        #[cfg(unix)]
        {
            let symlink = state.progress_file(&video()).unwrap();
            register_identity(&symlink, &video().id).unwrap();
            let external = directory.0.join("external.jsonl");
            append(
                &external,
                record(&video().id, "progress", true, 100.0).as_bytes(),
            );
            std::os::unix::fs::symlink(&external, &symlink).unwrap();
            assert!(state.poll_progress().is_err());
            assert!(!state.is_watched(&video().id));
            assert!(!state.poll_progress().unwrap());
        }
    }

    #[test]
    fn watching_requires_a_valid_positive_bound_acknowledgement() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        let report = state.progress_file(&video).unwrap();
        assert!(state.videos(CollectionKind::Watching).is_empty());
        assert!(!state.poll_progress().unwrap());
        register_identity(&report, &video.id).unwrap();
        assert!(!state.poll_progress().unwrap());
        for line in [
            record("rQ_J9WH6CGk", "progress", true, 100.0),
            record(&video.id, "progress", true, 0.0),
        ] {
            append(&report, line.as_bytes());
            assert!(!state.poll_progress().unwrap());
            assert!(state.videos(CollectionKind::Watching).is_empty());
        }
        for line in [
            record(&video.id, "unknown", true, 10.0),
            record(&video.id, "progress", true, -1.0),
            record(&video.id, "progress", true, 10.0)
                .replace("\"duration_seconds\":100", "\"duration_seconds\":0"),
            record(&video.id, "progress", true, 10.0).replace("\"version\":1", "\"version\":2"),
        ] {
            append(&report, line.as_bytes());
            assert!(state.poll_progress().is_err());
            assert!(state.videos(CollectionKind::Watching).is_empty());
            assert!(!state.is_watched(&video.id));
        }
        append(
            &report,
            record(&video.id, "progress", false, 10.0).as_bytes(),
        );
        assert!(state.poll_progress().unwrap());
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert!(!state.is_watched(&video.id));
    }

    #[test]
    fn starts_arriving_while_closed_preserve_metadata_without_a_creator() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let mut video = video();
        video.channel = None;
        video.channel_id = None;
        video.uploader_id = None;
        assert!(state.save(CollectionKind::Saved, &video).is_err());
        assert!(state.save(CollectionKind::WatchLater, &video).is_err());
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        drop(state);
        append(
            &report,
            record(&video.id, "closed", true, 10.0)
                .replace("\"duration_seconds\":100", "\"duration_seconds\":null")
                .as_bytes(),
        );
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.poll_progress().unwrap());
        let stored = &state.videos(CollectionKind::Watching)[0];
        assert_eq!(stored.id, video.id);
        assert_eq!(stored.title, video.title);
        assert_eq!(stored.channel, None);
        assert_eq!(stored.channel_id, None);
        assert_eq!(stored.uploader_id, None);
        assert_eq!(stored.publication_date, video.publication_date);
        assert_eq!(stored.timestamp, video.timestamp);
        append(&report, record(&video.id, "ended", false, 100.0).as_bytes());
        assert!(!state.poll_progress().unwrap());
        assert!(!state.is_watched(&video.id));
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert!(!state.poll_progress().unwrap());
    }

    #[test]
    fn completion_wins_over_parallel_sessions_late_reports_and_replays() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        for kind in [CollectionKind::Saved, CollectionKind::WatchLater] {
            state.save(kind, &video).unwrap();
        }
        let first = state.progress_file(&video).unwrap();
        register_identity(&first, &video.id).unwrap();
        append(&first, record(&video.id, "progress", true, 94.9).as_bytes());
        assert!(state.poll_progress().unwrap());
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert!(!state.is_watched(&video.id));
        let second = state.progress_file(&video).unwrap();
        register_identity(&second, &video.id).unwrap();
        append(
            &second,
            record(&video.id, "progress", true, 20.0).as_bytes(),
        );
        append(&first, record(&video.id, "ended", true, 95.0).as_bytes());
        assert!(state.poll_progress().unwrap());
        assert!(state.is_watched(&video.id));
        assert!(!state.contains(CollectionKind::Watching, &video.id));
        for report in [&first, &second] {
            append(report, record(&video.id, "progress", true, 30.0).as_bytes());
        }
        let replay = state.progress_file(&video).unwrap();
        register_identity(&replay, &video.id).unwrap();
        append(
            &replay,
            record(&video.id, "progress", true, 10.0).as_bytes(),
        );
        assert!(!state.poll_progress().unwrap());
        assert!(state.videos(CollectionKind::Watching).is_empty());
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(!state.poll_progress().unwrap());
        assert!(state.is_watched(&video.id));
        state
            .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
            .unwrap();
        assert!(state.is_watched(&video.id));
        assert!(state.contains(CollectionKind::Saved, &video.id));
        assert!(state.contains(CollectionKind::WatchLater, &video.id));
        assert!(state.videos(CollectionKind::Watching).is_empty());
    }

    #[test]
    fn manual_removal_suppresses_existing_sessions_across_restart_but_not_new_sessions() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        for kind in [CollectionKind::Saved, CollectionKind::WatchLater] {
            state.save(kind, &video).unwrap();
        }
        let first = state.progress_file(&video).unwrap();
        register_identity(&first, &video.id).unwrap();
        append(&first, record(&video.id, "progress", true, 10.0).as_bytes());
        assert!(state.poll_progress().unwrap());
        // Simulate a crash after collection persistence but before its admission marker.
        fs::remove_file(first.with_file_name("watching-seen.toml")).unwrap();
        let second = state.progress_file(&video).unwrap();
        register_identity(&second, &video.id).unwrap();
        append(
            &second,
            record(&video.id, "progress", true, 20.0).as_bytes(),
        );
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        state
            .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
            .unwrap();
        for report in [&first, &second] {
            append(report, record(&video.id, "progress", true, 50.0).as_bytes());
        }
        assert!(!state.poll_progress().unwrap());
        assert!(state.videos(CollectionKind::Watching).is_empty());
        assert!(!state.is_watched(&video.id));
        assert!(state.contains(CollectionKind::Saved, &video.id));
        assert!(state.contains(CollectionKind::WatchLater, &video.id));
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(!state.poll_progress().unwrap());
        let replay = state.progress_file(&video).unwrap();
        register_identity(&replay, &video.id).unwrap();
        append(
            &replay,
            record(&video.id, "progress", true, 10.0).as_bytes(),
        );
        assert!(state.poll_progress().unwrap());
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert!(!state.is_watched(&video.id));
    }

    #[test]
    fn watching_write_conflicts_retry_consumed_starts() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        let path = directory.0.join("currently-watching.toml");
        fs::write(&path, "external edit").unwrap();
        append(
            &report,
            record(&video.id, "progress", true, 10.0).as_bytes(),
        );
        assert!(state.poll_progress().is_err());
        assert!(state.videos(CollectionKind::Watching).is_empty());
        assert_eq!(fs::read_to_string(&path).unwrap(), "external edit");
        assert!(state.poll_progress().is_err());
        fs::remove_file(&path).unwrap();
        assert!(state.poll_progress().unwrap());
        assert!(state.contains(CollectionKind::Watching, &video.id));
        let original = fs::read(&path).unwrap();
        append(&path, b"\n# external removal conflict\n");
        let replay = state.progress_file(&video).unwrap();
        register_identity(&replay, &video.id).unwrap();
        append(
            &replay,
            record(&video.id, "progress", true, 20.0).as_bytes(),
        );
        assert!(state.poll_progress().is_err());
        assert!(!replay.with_file_name("watching-seen.toml").exists());
        fs::write(&path, original).unwrap();
        assert!(!state.poll_progress().unwrap());
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert!(!state.poll_progress().unwrap());
    }

    #[test]
    fn failed_session_admission_retries_without_losing_starts_or_allowing_readdition() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        state.streams[0].prepare().unwrap();
        let marker = report.with_file_name("watching-seen.toml");
        append(&marker, b"external edit");
        append(
            &report,
            record(&video.id, "progress", true, 10.0).as_bytes(),
        );
        assert!(state.poll_progress().is_err());
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert!(
            state
                .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
                .is_err()
        );
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert_eq!(fs::read_to_string(&marker).unwrap(), "external edit");
        fs::remove_file(&marker).unwrap();
        assert!(state.poll_progress().unwrap());
        state
            .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
            .unwrap();
        append(
            &report,
            record(&video.id, "progress", true, 50.0).as_bytes(),
        );
        assert!(!state.poll_progress().unwrap());
        assert!(state.videos(CollectionKind::Watching).is_empty());
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(!state.poll_progress().unwrap());
        assert!(state.videos(CollectionKind::Watching).is_empty());
    }

    #[test]
    fn failed_completion_cleanup_retries_and_restart_prunes_stale_watching() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        append(
            &report,
            record(&video.id, "progress", true, 10.0).as_bytes(),
        );
        assert!(state.poll_progress().unwrap());
        let path = directory.0.join("currently-watching.toml");
        let original = fs::read(&path).unwrap();
        append(&path, b"\n# external edit\n");
        let changed = fs::read(&path).unwrap();
        append(&report, record(&video.id, "ended", true, 95.0).as_bytes());
        assert!(state.poll_progress().is_err());
        assert!(state.is_watched(&video.id));
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert_eq!(fs::read(&path).unwrap(), changed);
        append(
            &report,
            record(&video.id, "progress", true, 20.0).as_bytes(),
        );
        assert!(state.poll_progress().is_err());
        fs::write(&path, original).unwrap();
        assert!(state.poll_progress().unwrap());
        assert!(!state.contains(CollectionKind::Watching, &video.id));
        drop(state);
        // Crash recovery must also handle a watched write committed before collection cleanup.
        fs::write(&path, video_document(std::iter::once(&video)).to_string()).unwrap();
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.is_watched(&video.id));
        assert!(state.videos(CollectionKind::Watching).is_empty());
        assert!(!state.poll_progress().unwrap());
    }

    #[test]
    fn legacy_sessions_use_saved_metadata_when_available_and_still_mark_completion_without_it() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        state.save(CollectionKind::Saved, &video).unwrap();
        let report = state.progress_file(&video).unwrap();
        fs::remove_file(report.with_file_name("video.toml")).unwrap();
        register_identity(&report, &video.id).unwrap();
        append(
            &report,
            record(&video.id, "progress", true, 10.0).as_bytes(),
        );
        assert!(state.poll_progress().unwrap());
        assert!(state.contains(CollectionKind::Watching, &video.id));
        append(&report, record(&video.id, "ended", true, 95.0).as_bytes());
        assert!(state.poll_progress().unwrap());
        assert!(state.is_watched(&video.id));
        assert!(state.videos(CollectionKind::Watching).is_empty());
        let mut other = video.clone();
        other.id = "rQ_J9WH6CGk".into();
        let report = state.progress_file(&other).unwrap();
        fs::remove_file(report.with_file_name("video.toml")).unwrap();
        register_identity(&report, &other.id).unwrap();
        append(
            &report,
            record(&other.id, "progress", true, 10.0).as_bytes(),
        );
        assert!(!state.poll_progress().unwrap());
        assert!(state.videos(CollectionKind::Watching).is_empty());
        append(&report, record(&other.id, "closed", true, 95.0).as_bytes());
        assert!(state.poll_progress().unwrap());
        assert!(state.is_watched(&other.id));
    }

    #[test]
    fn session_metadata_and_admission_markers_must_match_the_bound_identity() {
        for mismatched_snapshot in [true, false] {
            let directory = TestDirectory::new();
            let mut state = State::open(&directory.config()).unwrap();
            let video = video();
            let report = state.progress_file(&video).unwrap();
            register_identity(&report, &video.id).unwrap();
            if mismatched_snapshot {
                let mut other = video.clone();
                other.id = "rQ_J9WH6CGk".into();
                fs::write(
                    report.with_file_name("video.toml"),
                    video_document(std::iter::once(&other)).to_string(),
                )
                .unwrap();
            } else {
                append(
                    &report.with_file_name("watching-seen.toml"),
                    b"version = 1\nid = \"rQ_J9WH6CGk\"\n",
                );
            }
            append(&report, record(&video.id, "ended", true, 100.0).as_bytes());
            assert!(state.poll_progress().is_err());
            assert!(state.videos(CollectionKind::Watching).is_empty());
            assert!(!state.is_watched(&video.id));
        }
    }

    #[cfg(unix)]
    #[test]
    fn session_snapshots_reject_symlinks_and_shared_permissions() {
        use std::os::unix::fs::PermissionsExt;
        for symlink in [true, false] {
            let directory = TestDirectory::new();
            let mut state = State::open(&directory.config()).unwrap();
            let video = video();
            let report = state.progress_file(&video).unwrap();
            register_identity(&report, &video.id).unwrap();
            let snapshot = report.with_file_name("video.toml");
            if symlink {
                let external = directory.0.join("external.toml");
                fs::write(&external, fs::read(&snapshot).unwrap()).unwrap();
                fs::remove_file(&snapshot).unwrap();
                std::os::unix::fs::symlink(&external, &snapshot).unwrap();
            } else {
                fs::set_permissions(&snapshot, fs::Permissions::from_mode(0o644)).unwrap();
            }
            append(
                &report,
                record(&video.id, "progress", true, 100.0).as_bytes(),
            );
            assert!(state.poll_progress().is_err());
            assert!(state.videos(CollectionKind::Watching).is_empty());
            assert!(!state.is_watched(&video.id));
        }
    }

    #[test]
    fn watching_and_session_snapshots_keep_date_handle_channel_and_schema_validation() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        for invalid_field in ["date", "handle", "channel"] {
            let mut video = video();
            match invalid_field {
                "date" => video.publication_date = Some("2005-02-30".into()),
                "handle" => video.uploader_id = Some("not-a-handle".into()),
                "channel" => video.channel_id = Some("not-a-channel-id".into()),
                _ => unreachable!(),
            }
            assert!(state.progress_file(&video).is_err());
            assert!(state.save(CollectionKind::Watching, &video).is_err());
            assert!(state.videos(CollectionKind::Watching).is_empty());
        }
        let path = directory.0.join("currently-watching.toml");
        let contents = "version = 1\nvideos = []\nunexpected = true\n";
        fs::write(&path, contents).unwrap();
        assert!(State::open(&directory.config()).is_err());
        assert_eq!(fs::read_to_string(&path).unwrap(), contents);
    }

    #[test]
    fn watching_and_session_snapshots_reject_oversized_files() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        OpenOptions::new()
            .write(true)
            .open(report.with_file_name("video.toml"))
            .unwrap()
            .set_len(MAX_STATE_BYTES + 1)
            .unwrap();
        append(
            &report,
            record(&video.id, "progress", true, 100.0).as_bytes(),
        );
        assert!(state.poll_progress().is_err());
        assert!(state.videos(CollectionKind::Watching).is_empty());
        assert!(!state.is_watched(&video.id));
        File::create(directory.0.join("currently-watching.toml"))
            .unwrap()
            .set_len(MAX_STATE_BYTES + 1)
            .unwrap();
        assert!(State::open(&directory.config()).is_err());
    }

    #[test]
    fn miruro_sessions_recover_collections_and_watching_without_persisting_sources_in_state() {
        let directory = TestDirectory::new();
        let video = miruro_video(1);
        let mut state = State::open(&directory.config()).unwrap();
        for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
            state.save(kind, &video).unwrap();
        }
        let source = miruro_source(&video);
        let (source_file, report) = state.miruro_session(&video, &source).unwrap();
        let source_bytes = fs::read(&source_file).unwrap();
        assert_eq!(source_bytes, serde_json::to_vec(&source).unwrap());
        assert!(!report.exists());
        assert_eq!(
            fs::read_to_string(report.with_file_name("video-id")).unwrap(),
            video.id
        );
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            for path in [
                &source_file,
                &report.with_file_name("video-id"),
                &report.with_file_name("video.toml"),
            ] {
                assert_eq!(
                    fs::metadata(path).unwrap().permissions().mode() & 0o777,
                    0o600
                );
            }
            for path in [
                report.parent().unwrap(),
                &directory.0.join("playback-sessions"),
            ] {
                assert_eq!(
                    fs::metadata(path).unwrap().permissions().mode() & 0o777,
                    0o700
                );
            }
        }
        assert!(!state.poll_progress().unwrap());
        assert!(state.videos(CollectionKind::Watching).is_empty());
        drop(state);
        // A GUI can acknowledge its final snapshot after Tubular has quit.
        append(
            &report,
            miruro_record(&video, "closed", false, 10.0, None).as_bytes(),
        );
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.poll_progress().unwrap());
        let stored = &state.videos(CollectionKind::Watching)[0];
        assert_eq!(stored.id, video.id);
        assert_eq!(stored.title, video.title);
        assert_eq!(stored.channel, video.channel);
        assert_eq!(stored.timestamp, video.timestamp);
        assert_eq!(stored.publication_date, video.publication_date);
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.contains(CollectionKind::Watching, &video.id));
        assert!(!state.poll_progress().unwrap());
        append(
            &report,
            miruro_record(&video, "ended", true, 95.0, Some(100.0)).as_bytes(),
        );
        assert!(state.poll_progress().unwrap());
        assert!(state.is_watched(&video.id));
        assert!(state.videos(CollectionKind::Watching).is_empty());
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.is_watched(&video.id));
        assert!(!state.poll_progress().unwrap());
        for kind in [CollectionKind::WatchLater, CollectionKind::Saved] {
            let stored = &state.videos(kind)[0];
            assert_eq!(stored.id, video.id);
            assert_eq!(stored.title, video.title);
            assert_eq!(stored.channel, video.channel);
            assert_eq!(stored.channel_id, None);
            assert_eq!(stored.uploader_id, None);
            assert_eq!(stored.timestamp, video.timestamp);
            assert_eq!(stored.publication_date, video.publication_date);
        }
        for path in [
            directory.0.join("watch-later.toml"),
            directory.0.join("saved-videos.toml"),
            directory.0.join("currently-watching.toml"),
            directory.0.join("watch-state.toml"),
            report.with_file_name("video.toml"),
        ] {
            let contents = fs::read_to_string(path).unwrap();
            assert!(!contents.contains("private-token"));
            assert!(!contents.contains("cdn.example"));
        }
        assert_eq!(fs::read(source_file).unwrap(), source_bytes);
        assert!(report.exists());
    }

    #[test]
    fn miruro_v2_requires_matching_episode_eligible_known_duration_and_threshold() {
        for event in ["progress", "ended", "closed"] {
            let directory = TestDirectory::new();
            let mut state = State::open(&directory.config()).unwrap();
            let video = miruro_video(1);
            let (_, report) = state
                .miruro_session(&video, &miruro_source(&video))
                .unwrap();
            append(
                &report,
                miruro_record(&miruro_video(2), event, true, 100.0, Some(100.0)).as_bytes(),
            );
            assert!(!state.poll_progress().unwrap());
            assert!(state.videos(CollectionKind::Watching).is_empty());
            assert!(!state.is_watched(&miruro_video(2).id));
            for (eligible, position, duration) in [
                (true, 0.0, Some(100.0)),
                (true, 94.9, Some(100.0)),
                (false, 100.0, Some(100.0)),
                (true, 100.0, None),
            ] {
                append(
                    &report,
                    miruro_record(&video, event, eligible, position, duration).as_bytes(),
                );
                state.poll_progress().unwrap();
                assert!(!state.is_watched(&video.id));
            }
            assert!(state.contains(CollectionKind::Watching, &video.id));
            let completion = miruro_record(&video, event, true, 95.0, Some(100.0));
            append(&report, completion.as_bytes());
            append(&report, completion.as_bytes());
            assert!(state.poll_progress().unwrap());
            assert!(state.is_watched(&video.id));
            assert!(state.videos(CollectionKind::Watching).is_empty());
            assert!(!state.poll_progress().unwrap());
        }
    }

    #[test]
    fn miruro_v2_rejects_invalid_wire_schemas_numbers_and_namespaces() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = miruro_video(1);
        let (_, report) = state
            .miruro_session(&video, &miruro_source(&video))
            .unwrap();
        let valid = miruro_record(&video, "progress", true, 100.0, Some(100.0));
        for line in [
            valid.replace("\"version\":2", "\"version\":1"),
            valid.replace("\"version\":2", "\"version\":3"),
            valid.replace("\"provider\":\"miruro\"", "\"provider\":\"youtube\""),
            valid.replace("\"episode\":1", "\"episode\":0"),
            valid.replace("\"episode\":1", "\"episode\":2147483648"),
            valid.replace("\"position_seconds\":100.0", "\"position_seconds\":-1"),
            valid.replace("\"position_seconds\":100.0", "\"position_seconds\":1e999"),
            valid.replace("\"duration_seconds\":100.0", "\"duration_seconds\":0"),
            valid.replace("\"duration_seconds\":100.0", "\"duration_seconds\":-1"),
            valid.replace("\"duration_seconds\":100.0", "\"duration_seconds\":1e999"),
            valid.replace("\"event\":\"progress\"", "\"event\":\"other\""),
            valid.replacen('{', "{\"media_url\":null,", 1),
            record(&self::video().id, "ended", true, 100.0)
                .replace("\"version\":1", "\"version\":2"),
        ] {
            append(&report, line.as_bytes());
            assert!(state.poll_progress().is_err(), "{line}");
            assert!(state.videos(CollectionKind::Watching).is_empty());
            assert!(!state.is_watched(&video.id));
            assert!(!state.poll_progress().unwrap());
        }
    }

    #[test]
    fn miruro_partial_reports_and_bad_siblings_recover_after_restart() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = miruro_video(1);
        let (source, report) = state
            .miruro_session(&video, &miruro_source(&video))
            .unwrap();
        let completion = miruro_record(&video, "closed", true, 95.0, Some(100.0));
        append(&report, &completion.as_bytes()[..completion.len() - 1]);
        assert!(!state.poll_progress().unwrap());
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(!state.poll_progress().unwrap());
        let other = miruro_video(2);
        let (_, bad) = state
            .miruro_session(&other, &miruro_source(&other))
            .unwrap();
        append(&bad, b"not json\n");
        append(&report, b"\n");
        assert!(state.poll_progress().is_err());
        assert!(state.is_watched(&video.id));
        assert!(!state.is_watched(&other.id));
        assert!(state.poll_progress().unwrap());
        assert!(!state.poll_progress().unwrap());
        drop(state);
        assert!(source.exists());
        assert!(report.exists());
        let state = State::open(&directory.config()).unwrap();
        assert!(state.is_watched(&video.id));
    }

    #[test]
    fn miruro_state_rejects_catalog_rows_bad_titles_and_youtube_metadata() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        for field in [
            "series",
            "noncanonical",
            "empty-title",
            "control-title",
            "channel",
            "channel-id",
            "uploader",
        ] {
            let mut video = miruro_video(1);
            match field {
                "series" => video.id = "miruro-series:opaque_Show-1".into(),
                "noncanonical" => video.id = "miruro:opaque_Show-1:01".into(),
                "empty-title" => video.title = " ".into(),
                "control-title" => video.title.push('\n'),
                "channel" => video.channel = Some("YouTube".into()),
                "channel-id" => video.channel_id = self::video().channel_id,
                "uploader" => video.uploader_id = Some("@jawed".into()),
                _ => unreachable!(),
            }
            for kind in [
                CollectionKind::WatchLater,
                CollectionKind::Saved,
                CollectionKind::Watching,
            ] {
                assert!(state.save(kind, &video).is_err(), "{field}");
            }
            assert!(state.progress_file(&video).is_err(), "{field}");
        }
        for id in [
            "miruro-series:opaque_Show-1",
            "miruro:opaque_Show-1:01",
            "other:opaque_Show-1:1",
        ] {
            let mut watched = DocumentMut::new();
            watched["version"] = value(1);
            let mut ids = Array::new();
            ids.push(id);
            watched["watched"] = value(ids);
            assert!(load_watched(&watched).is_err());
        }
    }

    #[test]
    fn miruro_source_sessions_validate_binding_size_and_maximum_identity_marker() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let mut video = miruro_video(i32::MAX as u32);
        video.id = Identity::new("a".repeat(512), i32::MAX as u32)
            .unwrap()
            .key();
        let mut source = miruro_source(&video);
        source.identity.episode -= 1;
        assert!(state.miruro_session(&video, &source).is_err());
        assert!(state.streams.is_empty());
        source.identity = Identity::from_key(&video.id).unwrap();
        source.title = Some("x".repeat(MAX_SOURCE_BYTES));
        assert!(state.miruro_session(&video, &source).is_err());
        assert!(state.streams.is_empty());
        source.title = None;
        let (_, report) = state.miruro_session(&video, &source).unwrap();
        append(
            &report,
            miruro_record(&video, "ended", true, 95.0, Some(100.0)).as_bytes(),
        );
        drop(state);
        let mut state = State::open(&directory.config()).unwrap();
        assert!(state.poll_progress().unwrap());
        assert!(state.is_watched(&video.id));
        drop(state);
        let state = State::open(&directory.config()).unwrap();
        assert!(state.is_watched(&video.id));
    }

    #[test]
    fn miruro_recovery_rejects_missing_snapshots_and_overlong_identity_markers() {
        for missing_snapshot in [true, false] {
            let directory = TestDirectory::new();
            let mut state = State::open(&directory.config()).unwrap();
            let video = miruro_video(1);
            let (_, report) = state
                .miruro_session(&video, &miruro_source(&video))
                .unwrap();
            if missing_snapshot {
                fs::remove_file(report.with_file_name("video.toml")).unwrap();
            } else {
                append(
                    &report.with_file_name("video-id"),
                    &vec![b'x'; MAX_IDENTITY_BYTES as usize],
                );
            }
            append(
                &report,
                miruro_record(&video, "ended", true, 100.0, Some(100.0)).as_bytes(),
            );
            drop(state);
            let mut state = State::open(&directory.config()).unwrap();
            assert!(state.poll_progress().is_err());
            assert!(!state.is_watched(&video.id));
            assert!(state.videos(CollectionKind::Watching).is_empty());
        }
    }

    #[test]
    fn youtube_v1_reports_keep_canonical_urls_and_unknown_field_tolerance() {
        let directory = TestDirectory::new();
        let mut state = State::open(&directory.config()).unwrap();
        let video = video();
        let report = state.progress_file(&video).unwrap();
        register_identity(&report, &video.id).unwrap();
        let valid = record(&video.id, "ended", true, 95.0);
        for url in [
            format!("https://youtu.be/{}", video.id),
            format!("https://www.youtube.com/watch?v={}&t=1", video.id),
            format!("http://www.youtube.com/watch?v={}", video.id),
        ] {
            append(
                &report,
                valid
                    .replace(
                        &format!("https://www.youtube.com/watch?v={}", video.id),
                        &url,
                    )
                    .as_bytes(),
            );
            assert!(state.poll_progress().is_err());
            assert!(!state.is_watched(&video.id));
            assert!(state.videos(CollectionKind::Watching).is_empty());
        }
        append(
            &report,
            valid
                .replacen('{', "{\"legacy_extension\":true,", 1)
                .as_bytes(),
        );
        assert!(state.poll_progress().unwrap());
        assert!(state.is_watched(&video.id));
    }
}