arr/crates/arr-daemon/src/import.rs

//! The import pipeline: probe a completed download, judge it against the
//! policy a second time with real evidence, and link it into the library. See
//! DESIGN.md §5.7, §7.2, §7.3 and §7.4.
//!
//! The torrent's own files are never moved, renamed or deleted — the torrent
//! and the library entry are separate lifecycles (§7.3). A hard-failed
//! release is blacklisted and its grab marked failed, but the torrent keeps
//! seeding until Transmission's own limits clear it.
//!
//! Everything here is idempotent from domain rows (§8): a grab in
//! `downloaded` with no import recorded is the gap, and re-running any prefix
//! of the pipeline after a crash converges — the hardlink call tolerates the
//! destination already existing, and the `media_files` insert upserts on
//! path.

use std::collections::HashMap;
use std::ffi::OsString;
use std::io;
use std::path::{Component, Path, PathBuf};

use arr_core::layout;
use arr_core::policy::{evaluate, Candidate};
use arr_core::{ProbedMedia, Rule, Source, Verdict};
use arr_db::Db;
use arr_dl::TransmissionClient;
use arr_probe::Prober;

use crate::jellyfin::JellyfinClient;
use crate::notify::Notifier;
use crate::reconcile::{Action, ActionFuture, Outcome};

/// A failure during one import tick.
#[derive(Debug, thiserror::Error)]
pub enum ImportError {
    #[error("database: {0}")]
    Database(#[from] sqlx::Error),
    #[error("policy: {0}")]
    Policy(#[from] arr_db::PolicyError),
    #[error("transmission: {0}")]
    Transmission(#[from] arr_dl::Error),
    #[error("probe: {0}")]
    Probe(#[from] arr_probe::Error),
    #[error("blocking task: {0}")]
    Join(#[from] tokio::task::JoinError),
    #[error("{action} {path}: {source}")]
    Io {
        action: &'static str,
        path: PathBuf,
        source: io::Error,
    },
}

/// How the feature reached the library.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Placement {
    Linked,
    /// `link()` raised `EXDEV`; the file was copied instead (§7.2).
    Copied,
    /// The destination already existed — an earlier attempt placed it before
    /// the process died. Never a partial file: copies land via rename.
    AlreadyPlaced,
}

/// What one probe attempt settled about one path.
#[derive(Debug, Clone)]
enum ProbeOutcome {
    Media(Box<arr_probe::ProbedFile>),
    /// A fact about the file — `.nfo`, artwork, corrupt — not the prober.
    NotMedia,
}

/// Imports every downloaded grab: probe, second policy pass, hardlink into
/// the §7.4 layout.
#[derive(Debug)]
pub struct ImportAction {
    transmission: TransmissionClient,
    prober: Prober,
    jellyfin: JellyfinClient,
    notifier: Notifier,
    /// The operator's ntfy topic (DESIGN.md §9.5), for the *broken*
    /// notification a disk-full hardlink/copy failure raises. `None` when
    /// unconfigured: the failure is still logged, just not notified.
    operator_topic: Option<String>,
    /// Debounces the disk-full *broken* notification so a stuck-full disk
    /// notifies once, not every tick. Transient: a restart re-arms it, same
    /// as the probe cache above.
    disk_full_notified: std::sync::Arc<tokio::sync::Mutex<bool>>,
    /// Probe results by path, kept across ticks. The reconcile lane cancels
    /// the whole action after its 25 s budget while one probe alone may take
    /// up to 60 s, so without this a large multi-file torrent would restart
    /// from the first file every tick and never finish. Shared with the
    /// detached probe tasks, which is what lets a probe outlive a cancelled
    /// tick and still deposit its result. Transient state, rebuilt by
    /// re-probing after a restart (§8); entries are dropped once their grab
    /// settles.
    probed: std::sync::Arc<tokio::sync::Mutex<HashMap<PathBuf, ProbeOutcome>>>,
}

/// A movie grab Transmission finished downloading, not yet imported.
#[derive(Debug, Clone)]
struct PendingImport {
    grab_id: i64,
    infohash: String,
    movie_id: i64,
    tmdb_id: i64,
    title: String,
    year: Option<i64>,
    original_language: Option<String>,
    release_name: String,
}

impl ImportAction {
    #[must_use]
    pub fn new(
        transmission: TransmissionClient,
        prober: Prober,
        jellyfin: JellyfinClient,
        notifier: Notifier,
        operator_topic: Option<String>,
    ) -> Self {
        Self {
            transmission,
            prober,
            jellyfin,
            notifier,
            operator_topic,
            disk_full_notified: std::sync::Arc::new(tokio::sync::Mutex::new(false)),
            probed: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())),
        }
    }

    /// Probe every path, reusing results settled on earlier ticks, and
    /// return the readable video files.
    ///
    /// Each probe runs as a detached task that writes the cache itself, so a
    /// tick cancelled by the lane's 25 s budget mid-probe loses nothing: the
    /// probe finishes in the background (its own 60 s limit still applies)
    /// and the next tick reads the deposited result. A tick that re-requests
    /// a path an orphaned probe is still on starts a second probe of the same
    /// file — bounded overlap, identical result, chosen over tracking
    /// in-flight probes.
    ///
    /// Only facts about a file are cached; a prober failure — missing
    /// binary, timeout, unparseable output — is returned so the tick retries.
    async fn probe_all(
        &self,
        paths: &[PathBuf],
    ) -> Result<Vec<arr_probe::ProbedFile>, ImportError> {
        let mut files = Vec::new();
        for path in paths {
            let cached = self.probed.lock().await.get(path).cloned();
            let outcome = if let Some(outcome) = cached {
                outcome
            } else {
                let prober = self.prober.clone();
                let cache = std::sync::Arc::clone(&self.probed);
                let target = path.clone();
                tokio::spawn(async move {
                    let outcome = match prober.probe(target.clone()).await {
                        Ok(file) => ProbeOutcome::Media(Box::new(file)),
                        Err(error) if error.is_about_the_file() => {
                            tracing::debug!(path = %target.display(), %error, "not a video file, skipping");
                            ProbeOutcome::NotMedia
                        }
                        Err(error) => return Err(error),
                    };
                    cache.lock().await.insert(target, outcome.clone());
                    Ok(outcome)
                })
                .await??
            };
            if let ProbeOutcome::Media(file) = outcome {
                files.push(*file);
            }
        }
        Ok(files)
    }

    /// Drop a settled grab's probe results — imported or blacklisted, they
    /// will not be needed again.
    async fn forget_probes(&self, paths: &[PathBuf]) {
        let mut probed = self.probed.lock().await;
        for path in paths {
            probed.remove(path);
        }
    }

    async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, ImportError> {
        let mut outcomes = Vec::new();
        for pending in pending_imports(database).await? {
            match self.import_one(database, &pending).await {
                Ok(Some(outcome)) => {
                    self.clear_disk_full().await;
                    outcomes.push(outcome);
                }
                Ok(None) => {}
                // One grab's failure must not cost the rest of the tick.
                Err(error) => {
                    self.notify_if_disk_full(&error).await;
                    tracing::error!(
                        grab_id = pending.grab_id,
                        movie_id = pending.movie_id,
                        title = pending.title,
                        %error,
                        "import failed"
                    );
                }
            }
        }
        for pending in pending_tv_imports(database).await? {
            match self.import_tv_one(database, &pending).await {
                Ok(Some(outcome)) => {
                    self.clear_disk_full().await;
                    outcomes.push(outcome);
                }
                Ok(None) => {}
                Err(error) => {
                    self.notify_if_disk_full(&error).await;
                    tracing::error!(
                        grab_id = pending.grab_id,
                        series = pending.series_title,
                        %error,
                        "tv import failed"
                    );
                }
            }
        }
        Ok(outcomes)
    }

    /// §9.5 *broken*: a hardlink or copy that failed because the target
    /// filesystem is full. Debounced so a disk that stays full notifies once,
    /// not every tick, and silently re-arms once space frees up.
    async fn notify_if_disk_full(&self, error: &ImportError) {
        let ImportError::Io { source, .. } = error else {
            return;
        };
        if source.kind() != io::ErrorKind::StorageFull {
            return;
        }
        let Some(topic) = &self.operator_topic else {
            return;
        };
        let mut notified = self.disk_full_notified.lock().await;
        if *notified {
            return;
        }
        if let Err(notify_error) = self
            .notifier
            .send(topic, "arr: disk full", &error.to_string())
            .await
        {
            tracing::warn!(%notify_error, "broken notification failed");
            return;
        }
        *notified = true;
    }

    async fn clear_disk_full(&self) {
        *self.disk_full_notified.lock().await = false;
    }

    async fn import_one(
        &self,
        database: &Db,
        pending: &PendingImport,
    ) -> Result<Option<Outcome>, ImportError> {
        let Some(loaded) = database.movie_policy(pending.movie_id).await? else {
            return Ok(None);
        };
        // §5.2: no original language, nothing to judge audio against.
        let Some(original_language) = pending.original_language.as_deref() else {
            tracing::warn!(
                movie_id = pending.movie_id,
                title = pending.title,
                "no original language yet; not importing"
            );
            return Ok(None);
        };
        let original_language = arr_db::policy::language(original_language);

        let Some(paths) = self
            .torrent_paths(pending.grab_id, &pending.infohash)
            .await?
        else {
            return self.vanish(database, pending).await.map(Some);
        };

        // No expected runtime yet: the movies table carries no TMDB runtime,
        // so feature selection is by size alone (largest readable video).
        let mut candidates = self.probe_all(&paths).await?;
        candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.size));
        let Some(feature) = candidates.into_iter().next() else {
            // §5.7 "corrupt, wrong content": nothing in the torrent is a
            // readable video file.
            self.forget_probes(&paths).await;
            return self
                .hard_fail(database, pending, "no readable video file")
                .await
                .map(Some);
        };

        // §5.6 second phase of truth: same policy, real evidence.
        let evaluation = evaluate(
            &loaded.policy,
            &loaded.overrides,
            &original_language,
            Candidate::PostDownload(&feature.media),
            Some(feature.size),
            1,
            0,
        );
        let waiver: Option<Rule> = match evaluation.verdict {
            Verdict::Rejected(rule) => {
                self.forget_probes(&paths).await;
                return self
                    .hard_fail(database, pending, &rule.name())
                    .await
                    .map(Some);
            }
            Verdict::Waived(rule) => Some(rule),
            Verdict::Eligible => None,
        };

        // The source tag is the one claim a file cannot verify (§5.6); every
        // other tag comes from the probe.
        let claimed_source = arr_parse::parse(&pending.release_name)
            .source
            .map(Source::from);
        let tags = layout::attribute_tags(&feature.media, claimed_source);
        let extension = feature.path.extension().and_then(|ext| ext.to_str());
        let folder = layout::movie_folder(&pending.title, pending.year, pending.tmdb_id);
        let file_name = layout::movie_file_name(
            &pending.title,
            pending.year,
            pending.tmdb_id,
            &tags,
            extension,
        );
        let destination = Path::new(&loaded.root_path).join(folder).join(file_name);

        let source_path = feature.path.clone();
        let link_target = destination.clone();
        let placement =
            tokio::task::spawn_blocking(move || place(&source_path, &link_target)).await??;

        record_import(database, pending, &feature, waiver.as_ref(), &destination).await?;
        self.forget_probes(&paths).await;
        self.refresh_jellyfin().await;
        let path_text = destination.to_string_lossy().into_owned();
        tracing::info!(
            grab_id = pending.grab_id,
            movie_id = pending.movie_id,
            title = pending.title,
            path = path_text,
            placement = ?placement,
            waived = waiver.is_some(),
            "imported"
        );
        self.notify_imported(
            database,
            "movie",
            pending.movie_id,
            &title_with_year(&pending.title, pending.year),
        )
        .await;
        Ok(Some(Outcome::new(
            format!("grab {} downloaded, not imported", pending.grab_id),
            format!("imported {path_text}"),
        )))
    }

    /// §9.5 *imported*: the only good-news notification, sent to the title's
    /// owners alone. A failure to reach ntfy must not fail the import, which
    /// has already succeeded.
    async fn notify_imported(&self, database: &Db, title_kind: &str, title_id: i64, title: &str) {
        let topics = match owner_topics(database, title_kind, title_id).await {
            Ok(topics) => topics,
            Err(error) => {
                tracing::warn!(%error, "could not load owners for imported notification");
                return;
            }
        };
        for topic in topics {
            if let Err(error) = self.notifier.send(&topic, title, "imported").await {
                tracing::warn!(%error, topic, "imported notification failed");
            }
        }
    }

    /// The torrent's files as safe local paths, or `None` when Transmission
    /// no longer has the torrent.
    ///
    /// Torrent-declared names are untrusted input: an absolute or
    /// `..`-carrying entry would escape the download root and get probed —
    /// and possibly hardlinked — from anywhere on disk.
    async fn torrent_paths(
        &self,
        grab_id: i64,
        infohash: &str,
    ) -> Result<Option<Vec<PathBuf>>, ImportError> {
        let Some(content) = self.transmission.torrent_content(infohash).await? else {
            // Gone from Transmission — the caller marks the grab vanished
            // and parks the target (#108).
            tracing::warn!(
                grab_id,
                infohash,
                "downloaded grab has no torrent in Transmission; not importing"
            );
            return Ok(None);
        };
        Ok(Some(
            content
                .files
                .iter()
                .filter_map(|file| {
                    let path = safe_join(&content.download_dir, &file.path);
                    if path.is_none() {
                        tracing::warn!(
                            grab_id,
                            path = %file.path.display(),
                            "torrent file path escapes the download root; skipping"
                        );
                    }
                    path
                })
                .collect(),
        ))
    }

    /// Import one downloaded TV grab: a single episode or a season pack.
    ///
    /// A pack maps each video file to an episode by the `SxxEyy` tag in its
    /// own name, then imports the episodes that are missing. Episodes already
    /// on disk are skipped, never re-imported. If any mapped file fails the
    /// policy hard, the whole pack hard-fails: that release is blacklisted
    /// and the episodes reopen as gaps, which the grab selection then fills
    /// per episode rather than writing the season off.
    async fn import_tv_one(
        &self,
        database: &Db,
        pending: &PendingTvImport,
    ) -> Result<Option<Outcome>, ImportError> {
        // §5.2: no original language, nothing to judge audio against.
        let Some(original_language) = pending.original_language.as_deref() else {
            tracing::warn!(
                series = pending.series_title,
                "no original language yet; not importing"
            );
            return Ok(None);
        };
        let original_language = arr_db::policy::language(original_language);

        let episodes = target_episodes(database, pending).await?;
        let Some(first) = episodes.first() else {
            return Ok(None);
        };
        let Some(loaded) = database.episode_policy(first.id).await? else {
            return Ok(None);
        };

        let Some(paths) = self
            .torrent_paths(pending.grab_id, &pending.infohash)
            .await?
        else {
            return self.vanish_tv(database, pending).await.map(Some);
        };
        let candidates = self.probe_all(&paths).await?;
        let assignments = assign_files(pending, &episodes, candidates);
        if assignments.is_empty() {
            self.forget_probes(&paths).await;
            return self
                .hard_fail_tv(database, pending, "no file matches a wanted episode")
                .await
                .map(Some);
        }

        // §5.6 second phase of truth, over every file that would be
        // imported, before anything is placed: one hard failure condemns
        // the whole release (§5.7), not the episodes.
        let runtime_minutes = pending
            .runtime_minutes
            .and_then(|minutes| u32::try_from(minutes).ok())
            .unwrap_or(0);
        let mut imports = Vec::new();
        for assignment in assignments {
            if assignment.episode.has_file {
                // The partial-overlap case: this episode exists on disk and
                // is not re-imported, whatever the pack carries for it.
                tracing::info!(
                    grab_id = pending.grab_id,
                    episode_id = assignment.episode.id,
                    "episode already on disk; skipping its file in the pack"
                );
                continue;
            }
            let evaluation = evaluate(
                &loaded.policy,
                &loaded.overrides,
                &original_language,
                Candidate::PostDownload(&assignment.file.media),
                Some(assignment.file.size),
                1,
                runtime_minutes,
            );
            let waiver = match evaluation.verdict {
                Verdict::Rejected(rule) => {
                    self.forget_probes(&paths).await;
                    return self
                        .hard_fail_tv(database, pending, &rule.name())
                        .await
                        .map(Some);
                }
                Verdict::Waived(rule) => Some(rule),
                Verdict::Eligible => None,
            };
            imports.push((assignment, waiver));
        }
        if imports.is_empty() {
            // Everything the pack holds is already on disk. Nothing to
            // place; the grab is settled.
            mark_grab_imported(database, pending.grab_id).await?;
            self.forget_probes(&paths).await;
            return Ok(Some(Outcome::new(
                format!("grab {} downloaded, not imported", pending.grab_id),
                "every episode in the pack was already on disk".to_owned(),
            )));
        }

        let imported = self
            .place_episodes(database, pending, &loaded.root_path, imports)
            .await?;

        mark_grab_imported(database, pending.grab_id).await?;
        self.forget_probes(&paths).await;
        self.refresh_jellyfin().await;
        self.notify_imported(
            database,
            "series",
            pending.series_id,
            &title_with_year(&pending.series_title, pending.series_year),
        )
        .await;
        Ok(Some(Outcome::new(
            format!("grab {} downloaded, not imported", pending.grab_id),
            format!(
                "imported {imported} episode file(s) of {}",
                pending.series_title
            ),
        )))
    }

    /// Hardlink each judged file into the §7.4 TV layout and settle its rows.
    async fn place_episodes(
        &self,
        database: &Db,
        pending: &PendingTvImport,
        root_path: &str,
        imports: Vec<(Assignment, Option<Rule>)>,
    ) -> Result<usize, ImportError> {
        let claimed_source = arr_parse::parse(&pending.release_name)
            .source
            .map(Source::from);
        let season_number = u16::try_from(pending.season_number).unwrap_or_default();
        let mut imported = 0usize;
        for (assignment, waiver) in imports {
            let episode = &assignment.episode;
            let feature = &assignment.file;
            let tags = layout::attribute_tags(&feature.media, claimed_source);
            let extension = feature.path.extension().and_then(|ext| ext.to_str());
            let destination = Path::new(root_path)
                .join(layout::series_folder(
                    &pending.series_title,
                    pending.series_year,
                    pending.series_tmdb_id,
                ))
                .join(layout::season_folder(season_number))
                .join(layout::episode_file_name(
                    &pending.series_title,
                    pending.series_year,
                    season_number,
                    u16::try_from(episode.number).unwrap_or_default(),
                    &episode.title,
                    &tags,
                    extension,
                ));

            let source_path = feature.path.clone();
            let link_target = destination.clone();
            tokio::task::spawn_blocking(move || place(&source_path, &link_target)).await??;

            record_episode_import(database, episode.id, feature, waiver.as_ref(), &destination)
                .await?;
            imported += 1;
            tracing::info!(
                grab_id = pending.grab_id,
                episode_id = episode.id,
                series = pending.series_title,
                path = %destination.display(),
                waived = waiver.is_some(),
                "imported"
            );
        }
        Ok(imported)
    }

    /// §5.7 hard fail for a TV grab: blacklist the release, fail the grab and
    /// reopen only the episodes it was downloading. The season is never
    /// blacklisted — grab selection falls back to per-episode.
    async fn hard_fail_tv(
        &self,
        database: &Db,
        pending: &PendingTvImport,
        reason: &str,
    ) -> Result<Outcome, ImportError> {
        arr_db::blacklist::add(
            database.pool(),
            Some(&pending.infohash),
            &pending.release_name,
            reason,
        )
        .await?;
        sqlx::query!(
            "UPDATE grabs SET state = 'failed' WHERE id = ?",
            pending.grab_id
        )
        .execute(database.pool())
        .await?;
        match pending.episode_id {
            Some(episode_id) => {
                sqlx::query!(
                    "UPDATE episodes
                     SET    state = 'missing',
                            updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
                     WHERE  id = ?",
                    episode_id
                )
                .execute(database.pool())
                .await?;
            }
            None => {
                sqlx::query!(
                    "UPDATE episodes
                     SET    state = 'missing',
                            updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
                     WHERE  season_id = ? AND state = 'downloading'
                       AND  NOT EXISTS (
                                SELECT 1 FROM media_files f
                                WHERE  f.owner_kind = 'episode' AND f.owner_id = episodes.id
                            )",
                    pending.season_id
                )
                .execute(database.pool())
                .await?;
            }
        }

        tracing::warn!(
            grab_id = pending.grab_id,
            series = pending.series_title,
            release = pending.release_name,
            reason,
            "hard fail post-probe; release blacklisted, episodes reopened, torrent left seeding"
        );
        Ok(Outcome::new(
            format!("grab {} hard-failed post-probe: {reason}", pending.grab_id),
            format!("blacklisted {}", pending.release_name),
        ))
    }

    /// §86/#108: a `downloaded` grab whose torrent Transmission no longer
    /// reports — removed by hand, not a policy failure. Marked `vanished`
    /// rather than `failed` so it does not feed the `needs_decision` queue
    /// (attention.rs). Nothing is blacklisted, since the release itself
    /// never failed policy, and the movie is parked rather than reopened,
    /// since removing a torrent by hand is human intent, not a gap to
    /// refill.
    async fn vanish(&self, database: &Db, pending: &PendingImport) -> Result<Outcome, ImportError> {
        sqlx::query!(
            "UPDATE grabs SET state = 'vanished' WHERE id = ?",
            pending.grab_id
        )
        .execute(database.pool())
        .await?;
        crate::grab::park_target(database, "movie", pending.movie_id).await?;

        tracing::warn!(
            grab_id = pending.grab_id,
            movie_id = pending.movie_id,
            title = pending.title,
            release = pending.release_name,
            "torrent vanished from Transmission; movie parked"
        );
        Ok(Outcome::new(
            format!(
                "grab {} downloaded, torrent vanished from Transmission",
                pending.grab_id
            ),
            format!("parked movie {}", pending.movie_id),
        ))
    }

    /// TV counterpart of [`Self::vanish`]: parks the episode, or the
    /// still-downloading episodes of a season pack.
    async fn vanish_tv(
        &self,
        database: &Db,
        pending: &PendingTvImport,
    ) -> Result<Outcome, ImportError> {
        sqlx::query!(
            "UPDATE grabs SET state = 'vanished' WHERE id = ?",
            pending.grab_id
        )
        .execute(database.pool())
        .await?;
        let (target_kind, target_id) = match pending.episode_id {
            Some(episode_id) => ("episode", episode_id),
            None => ("season", pending.season_id),
        };
        crate::grab::park_target(database, target_kind, target_id).await?;

        tracing::warn!(
            grab_id = pending.grab_id,
            series = pending.series_title,
            release = pending.release_name,
            "torrent vanished from Transmission; target parked"
        );
        Ok(Outcome::new(
            format!(
                "grab {} downloaded, torrent vanished from Transmission",
                pending.grab_id
            ),
            format!("parked {target_kind} {target_id}"),
        ))
    }

    /// §7.5: the filesystem watcher misses the just-hardlinked file. A
    /// failure to reach Jellyfin must not fail the import, which has already
    /// succeeded.
    async fn refresh_jellyfin(&self) {
        if let Err(error) = self.jellyfin.refresh().await {
            tracing::warn!(%error, "jellyfin refresh failed");
        }
    }

    /// §5.7 hard fail: blacklist the release, fail the grab, reopen the gap.
    /// The torrent is deliberately untouched (§7.3).
    async fn hard_fail(
        &self,
        database: &Db,
        pending: &PendingImport,
        reason: &str,
    ) -> Result<Outcome, ImportError> {
        arr_db::blacklist::add(
            database.pool(),
            Some(&pending.infohash),
            &pending.release_name,
            reason,
        )
        .await?;
        sqlx::query!(
            "UPDATE grabs SET state = 'failed' WHERE id = ?",
            pending.grab_id
        )
        .execute(database.pool())
        .await?;
        sqlx::query!(
            "UPDATE movies
             SET    state = 'missing',
                    updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
             WHERE  id = ?",
            pending.movie_id
        )
        .execute(database.pool())
        .await?;

        tracing::warn!(
            grab_id = pending.grab_id,
            movie_id = pending.movie_id,
            title = pending.title,
            release = pending.release_name,
            reason,
            "hard fail post-probe; blacklisted, torrent left seeding"
        );
        Ok(Outcome::new(
            format!("grab {} hard-failed post-probe: {reason}", pending.grab_id),
            format!("blacklisted {}", pending.release_name),
        ))
    }
}

impl Action for ImportAction {
    fn name(&self) -> &'static str {
        "import"
    }

    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
        Box::pin(async move { self.tick(database).await.map_err(Into::into) })
    }
}

/// Settle a TV grab as imported, whether or not any file was placed — a
/// pack entirely already on disk still needs its grab marked done.
async fn mark_grab_imported(database: &Db, grab_id: i64) -> Result<(), ImportError> {
    sqlx::query!(
        "UPDATE grabs
         SET    state = 'imported',
                imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
         WHERE  id = ?",
        grab_id
    )
    .execute(database.pool())
    .await?;
    Ok(())
}

/// A notification title: the bare title, or with the release year appended.
fn title_with_year(title: &str, year: Option<i64>) -> String {
    match year {
        Some(year) => format!("{title} ({year})"),
        None => title.to_string(),
    }
}

/// The ntfy topics of a title's owners (§4.3, §9.5), movie or series alike.
async fn owner_topics(
    database: &Db,
    title_kind: &str,
    title_id: i64,
) -> Result<Vec<String>, ImportError> {
    Ok(sqlx::query_scalar!(
        r#"SELECT o.ntfy_topic AS "ntfy_topic!: String"
           FROM   owners o
           JOIN   title_owners t ON t.owner_id = o.id
           WHERE  t.title_kind = ? AND t.title_id = ?"#,
        title_kind,
        title_id
    )
    .fetch_all(database.pool())
    .await?)
}

/// Settle a placed file into the rows: the `media_files` record (§4), the
/// grab and the movie. The upsert on path is the crash seam — a re-run after
/// a death between the link and here converges instead of erroring.
async fn record_import(
    database: &Db,
    pending: &PendingImport,
    feature: &arr_probe::ProbedFile,
    waiver: Option<&Rule>,
    destination: &Path,
) -> Result<(), ImportError> {
    let probed = probed_json(&feature.media).to_string();
    let waiver_json = waiver.map(|rule| serde_json::json!({ "rule": rule.name() }).to_string());
    let size = i64::try_from(feature.size).unwrap_or(i64::MAX);
    let path_text = destination.to_string_lossy().into_owned();
    sqlx::query!(
        "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)
         VALUES ('movie', ?, ?, ?, ?, ?)
         ON CONFLICT (path) DO UPDATE SET
             size       = excluded.size,
             probed     = excluded.probed,
             waiver     = excluded.waiver,
             updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
        pending.movie_id,
        path_text,
        size,
        probed,
        waiver_json
    )
    .execute(database.pool())
    .await?;
    sqlx::query!(
        "UPDATE grabs
         SET    state = 'imported',
                imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
         WHERE  id = ?",
        pending.grab_id
    )
    .execute(database.pool())
    .await?;
    sqlx::query!(
        "UPDATE movies
         SET    state = 'available',
                updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
         WHERE  id = ?",
        pending.movie_id
    )
    .execute(database.pool())
    .await?;
    Ok(())
}

/// The gap, straight out of the domain rows (§8): a movie grab Transmission
/// finished that no import has settled.
async fn pending_imports(database: &Db) -> Result<Vec<PendingImport>, ImportError> {
    let rows = sqlx::query!(
        r#"
        SELECT   g.id                AS "grab_id!: i64",
                 g.infohash          AS "infohash!: String",
                 m.id                AS "movie_id!: i64",
                 m.tmdb_id           AS "tmdb_id!: i64",
                 m.title             AS "title!: String",
                 m.year,
                 m.original_language,
                 r.name              AS "release_name!: String"
        FROM     grabs g
        JOIN     movies m   ON m.id = g.target_id
        JOIN     releases r ON r.id = g.release_id
        WHERE    g.state = 'downloaded' AND g.target_kind = 'movie'
        ORDER BY g.id
        "#
    )
    .fetch_all(database.pool())
    .await?;

    Ok(rows
        .into_iter()
        .map(|row| PendingImport {
            grab_id: row.grab_id,
            infohash: row.infohash,
            movie_id: row.movie_id,
            tmdb_id: row.tmdb_id,
            title: row.title,
            year: row.year,
            original_language: row.original_language,
            release_name: row.release_name,
        })
        .collect())
}

/// A TV grab Transmission finished downloading — one episode or a season
/// pack — not yet imported.
#[derive(Debug, Clone)]
struct PendingTvImport {
    grab_id: i64,
    infohash: String,
    /// `Some` for an episode grab, `None` for a season pack.
    episode_id: Option<i64>,
    season_id: i64,
    season_number: i64,
    series_id: i64,
    series_tmdb_id: i64,
    series_title: String,
    series_year: Option<i64>,
    original_language: Option<String>,
    /// §5.5: the series' minutes per episode, scaling the size bands the
    /// same way the pre-grab verdict scaled them. `None` applies them
    /// unscaled.
    runtime_minutes: Option<i64>,
    release_name: String,
}

/// An episode a downloaded TV grab could satisfy.
#[derive(Debug, Clone)]
struct TargetEpisode {
    id: i64,
    number: i64,
    title: String,
    has_file: bool,
}

/// One probed video file tied to the episode it holds.
#[derive(Debug)]
struct Assignment {
    episode: TargetEpisode,
    file: arr_probe::ProbedFile,
}

/// The TV side of the gap (§8): downloaded episode and season grabs that no
/// import has settled.
async fn pending_tv_imports(database: &Db) -> Result<Vec<PendingTvImport>, ImportError> {
    let mut pending = Vec::new();

    let episode_rows = sqlx::query!(
        r#"
        SELECT   g.id       AS "grab_id!: i64",
                 g.infohash AS "infohash!: String",
                 e.id       AS "episode_id!: i64",
                 se.id      AS "season_id!: i64",
                 se.number  AS "season_number!: i64",
                 s.id       AS "series_id!: i64",
                 s.tmdb_id  AS "series_tmdb_id!: i64",
                 s.title    AS "series_title!: String",
                 s.year     AS "series_year",
                 s.original_language,
                 s.runtime_minutes,
                 r.name     AS "release_name!: String"
        FROM     grabs g
        JOIN     episodes e ON e.id = g.target_id
        JOIN     seasons se ON se.id = e.season_id
        JOIN     series s   ON s.id = se.series_id
        JOIN     releases r ON r.id = g.release_id
        WHERE    g.state = 'downloaded' AND g.target_kind = 'episode'
        ORDER BY g.id
        "#
    )
    .fetch_all(database.pool())
    .await?;
    pending.extend(episode_rows.into_iter().map(|row| PendingTvImport {
        grab_id: row.grab_id,
        infohash: row.infohash,
        episode_id: Some(row.episode_id),
        season_id: row.season_id,
        season_number: row.season_number,
        series_id: row.series_id,
        series_tmdb_id: row.series_tmdb_id,
        series_title: row.series_title,
        series_year: row.series_year,
        original_language: row.original_language,
        runtime_minutes: row.runtime_minutes,
        release_name: row.release_name,
    }));

    let season_rows = sqlx::query!(
        r#"
        SELECT   g.id       AS "grab_id!: i64",
                 g.infohash AS "infohash!: String",
                 se.id      AS "season_id!: i64",
                 se.number  AS "season_number!: i64",
                 s.id       AS "series_id!: i64",
                 s.tmdb_id  AS "series_tmdb_id!: i64",
                 s.title    AS "series_title!: String",
                 s.year     AS "series_year",
                 s.original_language,
                 s.runtime_minutes,
                 r.name     AS "release_name!: String"
        FROM     grabs g
        JOIN     seasons se ON se.id = g.target_id
        JOIN     series s   ON s.id = se.series_id
        JOIN     releases r ON r.id = g.release_id
        WHERE    g.state = 'downloaded' AND g.target_kind = 'season'
        ORDER BY g.id
        "#
    )
    .fetch_all(database.pool())
    .await?;
    pending.extend(season_rows.into_iter().map(|row| PendingTvImport {
        grab_id: row.grab_id,
        infohash: row.infohash,
        episode_id: None,
        season_id: row.season_id,
        season_number: row.season_number,
        series_id: row.series_id,
        series_tmdb_id: row.series_tmdb_id,
        series_title: row.series_title,
        series_year: row.series_year,
        original_language: row.original_language,
        runtime_minutes: row.runtime_minutes,
        release_name: row.release_name,
    }));

    pending.sort_by_key(|row| row.grab_id);
    Ok(pending)
}

/// The episodes a grab could satisfy: one for an episode grab, the whole
/// season for a pack.
async fn target_episodes(
    database: &Db,
    pending: &PendingTvImport,
) -> Result<Vec<TargetEpisode>, ImportError> {
    let rows = sqlx::query!(
        r#"
        SELECT   e.id     AS "id!: i64",
                 e.number AS "number!: i64",
                 e.title  AS "title!: String",
                 EXISTS (
                     SELECT 1 FROM media_files f
                     WHERE  f.owner_kind = 'episode' AND f.owner_id = e.id
                 ) AS "has_file!: bool"
        FROM     episodes e
        WHERE    e.season_id = ?
        ORDER BY e.number
        "#,
        pending.season_id
    )
    .fetch_all(database.pool())
    .await?;
    let episodes = rows.into_iter().map(|row| TargetEpisode {
        id: row.id,
        number: row.number,
        title: row.title,
        has_file: row.has_file,
    });
    Ok(match pending.episode_id {
        Some(episode_id) => episodes
            .filter(|episode| episode.id == episode_id)
            .collect(),
        None => episodes.collect(),
    })
}

/// Tie each readable video file to the episode its own name claims (§5.6:
/// per-file names are the only pre-probe truth a pack carries).
///
/// A file claiming several episodes lands on the first target it covers, one
/// file per episode, largest file winning a collision. For a single-episode
/// grab whose only video file carries no tag, the file is the episode.
fn assign_files(
    pending: &PendingTvImport,
    episodes: &[TargetEpisode],
    files: Vec<arr_probe::ProbedFile>,
) -> Vec<Assignment> {
    let season = u32::try_from(pending.season_number).unwrap_or_default();
    let mut by_episode: HashMap<i64, arr_probe::ProbedFile> = HashMap::new();
    let mut untagged: Vec<arr_probe::ProbedFile> = Vec::new();

    for file in files {
        let name = file.path.file_name().and_then(|name| name.to_str());
        let claim = name.and_then(|name| arr_parse::parse(name).episode);
        let Some(claim) = claim else {
            untagged.push(file);
            continue;
        };
        let covered = episodes.iter().find(|episode| {
            claim.covers(season, u32::try_from(episode.number).unwrap_or_default())
        });
        let Some(episode) = covered else {
            continue;
        };
        match by_episode.entry(episode.id) {
            std::collections::hash_map::Entry::Vacant(entry) => {
                entry.insert(file);
            }
            std::collections::hash_map::Entry::Occupied(mut entry) => {
                if file.size > entry.get().size {
                    entry.insert(file);
                }
            }
        }
    }

    // A single-episode torrent often names its one file after nothing
    // useful. One target, one untagged video: that is the episode.
    if pending.episode_id.is_some() && by_episode.is_empty() && untagged.len() == 1 {
        if let (Some(episode), Some(file)) = (episodes.first(), untagged.pop()) {
            by_episode.insert(episode.id, file);
        }
    }

    let mut assignments: Vec<Assignment> = episodes
        .iter()
        .filter_map(|episode| {
            by_episode.remove(&episode.id).map(|file| Assignment {
                episode: episode.clone(),
                file,
            })
        })
        .collect();
    assignments.sort_by_key(|assignment| assignment.episode.number);
    assignments
}

/// Settle a placed episode file into the rows: the `media_files` record, and
/// the episode itself. The upsert on path is the same crash seam the movie
/// import leans on.
async fn record_episode_import(
    database: &Db,
    episode_id: i64,
    feature: &arr_probe::ProbedFile,
    waiver: Option<&Rule>,
    destination: &Path,
) -> Result<(), ImportError> {
    let probed = probed_json(&feature.media).to_string();
    let waiver_json = waiver.map(|rule| serde_json::json!({ "rule": rule.name() }).to_string());
    let size = i64::try_from(feature.size).unwrap_or(i64::MAX);
    let path_text = destination.to_string_lossy().into_owned();
    sqlx::query!(
        "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)
         VALUES ('episode', ?, ?, ?, ?, ?)
         ON CONFLICT (path) DO UPDATE SET
             size       = excluded.size,
             probed     = excluded.probed,
             waiver     = excluded.waiver,
             updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
        episode_id,
        path_text,
        size,
        probed,
        waiver_json
    )
    .execute(database.pool())
    .await?;
    sqlx::query!(
        "UPDATE episodes
         SET    state = 'available',
                updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
         WHERE  id = ?",
        episode_id
    )
    .execute(database.pool())
    .await?;
    Ok(())
}

/// The `probed` column (§4, §5.6): what `ffprobe` found, in the spellings the
/// policy columns use.
fn probed_json(media: &ProbedMedia) -> serde_json::Value {
    serde_json::json!({
        "resolution": media.resolution.to_string(),
        "source": media.source.map(|source| source.to_string()),
        "hdr": media.hdr.to_string(),
        "audio_tracks": media
            .audio_tracks
            .iter()
            .map(|track| serde_json::json!({
                "language": track.language.to_string(),
                "title": track.title,
                "handler_name": track.handler_name,
            }))
            .collect::<Vec<_>>(),
        "sub_tracks": media
            .subtitle_tracks
            .iter()
            .map(|track| serde_json::json!({ "language": track.language.to_string() }))
            .collect::<Vec<_>>(),
    })
}

/// Join a torrent-declared file path onto the download root, refusing
/// anything that could land outside it: absolute paths, drive prefixes and
/// `..` components. `None` means the entry is hostile or malformed.
fn safe_join(root: &Path, declared: &Path) -> Option<PathBuf> {
    let mut clean = PathBuf::new();
    for component in declared.components() {
        match component {
            Component::Normal(part) => clean.push(part),
            Component::CurDir => {}
            Component::RootDir | Component::Prefix(_) | Component::ParentDir => return None,
        }
    }
    if clean.as_os_str().is_empty() {
        return None;
    }
    Some(root.join(clean))
}

/// Hardlink `source` to `destination`, falling back to copy on `EXDEV` only
/// (§7.2). No configuration flag.
fn place(source: &Path, destination: &Path) -> Result<Placement, ImportError> {
    if let Some(parent) = destination.parent() {
        std::fs::create_dir_all(parent).map_err(|error| ImportError::Io {
            action: "create library folder",
            path: parent.to_path_buf(),
            source: error,
        })?;
    }

    match std::fs::hard_link(source, destination) {
        Ok(()) => Ok(Placement::Linked),
        // A completed earlier attempt: links and copies both land whole
        // (copies via rename), so an existing destination is a finished one.
        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => Ok(Placement::AlreadyPlaced),
        Err(error) if error.kind() == io::ErrorKind::CrossesDevices => {
            copy_into_place(source, destination)
        }
        Err(error) => Err(ImportError::Io {
            action: "hardlink into",
            path: destination.to_path_buf(),
            source: error,
        }),
    }
}

/// Copy through a dot-name in the destination folder, then rename, so the
/// library never shows a partial file.
fn copy_into_place(source: &Path, destination: &Path) -> Result<Placement, ImportError> {
    let mut temp_name = OsString::from(".");
    temp_name.push(destination.file_name().unwrap_or_default());
    temp_name.push(".partial");
    let temp = destination.with_file_name(temp_name);

    let copied = std::fs::copy(source, &temp).and_then(|_| std::fs::rename(&temp, destination));
    if let Err(error) = copied {
        let _ = std::fs::remove_file(&temp);
        return Err(ImportError::Io {
            action: "copy into",
            path: destination.to_path_buf(),
            source: error,
        });
    }
    Ok(Placement::Copied)
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use std::os::unix::fs::{MetadataExt, PermissionsExt};
    use std::path::PathBuf;

    use serde_json::json;
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    use super::*;

    const INFOHASH: &str = "0123456789abcdef0123456789abcdef01234567";
    const RELEASE_NAME: &str = "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos";

    /// A 2160p HDR10 file with an English track — what the seeded main-movies
    /// policy accepts. The size is the container's claim, matching §5.5's
    /// band; the bytes on disk are tiny.
    const HDR10_PROBE: &str = r#"{
        "format": {"format_name": "matroska,webm", "duration": "9060.0", "size": "23622320128"},
        "streams": [
            {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600,
             "color_transfer": "smpte2084"},
            {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}}
        ]
    }"#;

    /// The same file as a Dolby Vision Profile 5 stream — §5.3's hard reject.
    const DV5_PROBE: &str = r#"{
        "format": {"format_name": "matroska,webm", "duration": "9060.0", "size": "23622320128"},
        "streams": [
            {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600,
             "color_transfer": "smpte2084",
             "side_data_list": [{"side_data_type": "DOVI configuration record", "dv_profile": 5}]},
            {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}}
        ]
    }"#;

    /// Tags a title with one owner, so a test can assert an *imported*
    /// notification reaches that owner's topic alone (§9.5).
    async fn insert_owner(database: &Db, title_kind: &str, title_id: i64, name: &str, topic: &str) {
        let owner_id = sqlx::query("INSERT INTO owners (name, ntfy_topic) VALUES (?, ?)")
            .bind(name)
            .bind(topic)
            .execute(database.pool())
            .await
            .unwrap()
            .last_insert_rowid();
        sqlx::query("INSERT INTO title_owners (title_kind, title_id, owner_id) VALUES (?, ?, ?)")
            .bind(title_kind)
            .bind(title_id)
            .bind(owner_id)
            .execute(database.pool())
            .await
            .unwrap();
    }

    async fn start_ntfy_server() -> MockServer {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;
        server
    }

    async fn start_jellyfin_server() -> MockServer {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/Library/Refresh"))
            .respond_with(ResponseTemplate::new(204))
            .mount(&server)
            .await;
        server
    }

    struct Harness {
        _dir: tempfile::TempDir,
        database: Db,
        downloads: PathBuf,
        library: PathBuf,
        action: ImportAction,
        _server: MockServer,
        jellyfin_server: MockServer,
        ntfy_server: MockServer,
    }

    /// An `ffprobe` stand-in: canned JSON for media, a `tty` document for the
    /// `.nfo`, so feature selection sees what the real binary would report.
    fn fake_ffprobe(directory: &Path, media_json: &str) -> PathBuf {
        let path = directory.join("ffprobe");
        let script = format!(
            "#!/bin/sh\nfor arg; do last=\"$arg\"; done\ncase \"$last\" in\n  *.nfo) printf '%s' '{{\"format\":{{\"format_name\":\"tty\"}}}}' ;;\n  *) cat <<'PROBE_EOF'\n{media_json}\nPROBE_EOF\n;;\nesac\n"
        );
        std::fs::write(&path, script).unwrap();
        let mut permissions = std::fs::metadata(&path).unwrap().permissions();
        permissions.set_mode(0o755);
        std::fs::set_permissions(&path, permissions).unwrap();
        path
    }

    async fn harness(media_json: &str) -> Harness {
        harness_with(
            media_json,
            json!([
                {"name": "Dune/Dune.mkv", "length": 13, "bytesCompleted": 13},
                {"name": "Dune/Dune.nfo", "length": 10, "bytesCompleted": 10}
            ]),
        )
        .await
    }

    async fn harness_with(media_json: &str, files: serde_json::Value) -> Harness {
        let dir = tempfile::tempdir().unwrap();
        let downloads = dir.path().join("downloads");
        let library = dir.path().join("library");
        std::fs::create_dir_all(downloads.join("Dune")).unwrap();
        std::fs::create_dir_all(&library).unwrap();
        std::fs::write(downloads.join("Dune/Dune.mkv"), b"feature bytes").unwrap();
        std::fs::write(downloads.join("Dune/Dune.nfo"), b"not a film").unwrap();

        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
        database.migrate().await.unwrap();
        let library_text = library.to_string_lossy().into_owned();
        sqlx::query("UPDATE roots SET path = ? WHERE kind = 'movie' AND audience = 'main'")
            .bind(&library_text)
            .execute(database.pool())
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, state)
             SELECT 693134, 'Dune: Part Two', 2024, 'en', id, 'downloading'
             FROM roots WHERE kind = 'movie' AND audience = 'main'",
        )
        .execute(database.pool())
        .await
        .unwrap();
        let release_id = sqlx::query(
            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
             VALUES (7, 'good', ?, 23622320128, 'magnet:x', '{}', 'eligible')",
        )
        .bind(RELEASE_NAME)
        .execute(database.pool())
        .await
        .unwrap()
        .last_insert_rowid();
        sqlx::query(
            "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
             VALUES (?, 'movie', 1, ?, 'downloaded')",
        )
        .bind(release_id)
        .bind(INFOHASH)
        .execute(database.pool())
        .await
        .unwrap();
        insert_owner(&database, "movie", 1, "Alice", "alice-topic").await;

        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "result": "success",
                "arguments": {"torrents": [{
                    "hashString": INFOHASH,
                    "downloadDir": downloads.to_string_lossy(),
                    "files": files
                }]}
            })))
            .mount(&server)
            .await;

        let jellyfin_server = start_jellyfin_server().await;
        let ntfy_server = start_ntfy_server().await;

        let prober = Prober::new().with_binary(fake_ffprobe(dir.path(), media_json));
        let jellyfin = JellyfinClient::new(jellyfin_server.uri(), None).unwrap();
        let notifier = Notifier::new(ntfy_server.uri()).unwrap();
        let action = ImportAction::new(
            TransmissionClient::new(&server.uri()).unwrap(),
            prober,
            jellyfin,
            notifier,
            Some("operator-topic".to_string()),
        );

        Harness {
            _dir: dir,
            database,
            downloads,
            library,
            action,
            _server: server,
            jellyfin_server,
            ntfy_server,
        }
    }

    fn expected_library_file(library: &Path) -> PathBuf {
        library
            .join("Dune Part Two (2024) [tmdbid-693134]")
            .join("Dune Part Two (2024) [tmdbid-693134] - [2160p][WEB-DL][HDR10].mkv")
    }

    /// The issue's acceptance case: the exact §7.4 path exists, and the
    /// torrent's own file still exists with a link count of two.
    #[tokio::test]
    async fn the_feature_lands_on_the_design_layout_and_keeps_seeding() {
        let h = harness(HDR10_PROBE).await;

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        let library_file = expected_library_file(&h.library);
        assert!(library_file.is_file(), "missing {}", library_file.display());

        let seeding_file = h.downloads.join("Dune/Dune.mkv");
        let metadata = std::fs::metadata(&seeding_file).unwrap();
        assert_eq!(metadata.nlink(), 2, "§7.2: hardlinked, not moved or copied");

        let (path, probed, waiver): (String, String, Option<String>) =
            sqlx::query_as("SELECT path, probed, waiver FROM media_files WHERE owner_kind = 'movie' AND owner_id = 1")
                .fetch_one(h.database.pool())
                .await
                .unwrap();
        assert_eq!(path, library_file.to_string_lossy());
        assert!(probed.contains("\"2160p\""), "{probed}");
        assert!(probed.contains("HDR10"), "{probed}");
        assert_eq!(waiver, None);

        let (grab_state, imported_at): (String, Option<String>) =
            sqlx::query_as("SELECT state, imported_at FROM grabs WHERE infohash = ?")
                .bind(INFOHASH)
                .fetch_one(h.database.pool())
                .await
                .unwrap();
        assert_eq!(grab_state, "imported");
        assert!(imported_at.is_some());
        let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(movie_state, "available");

        assert_eq!(
            h.jellyfin_server.received_requests().await.unwrap().len(),
            1,
            "§7.5: one refresh call at the end of a successful import"
        );

        // The issue's acceptance case: an import notifies only that title's
        // owners (§9.5), never the operator topic.
        let notifications = h.ntfy_server.received_requests().await.unwrap();
        assert_eq!(notifications.len(), 1);
        assert_eq!(notifications[0].url.path(), "/alice-topic");
        let body = String::from_utf8(notifications[0].body.clone()).unwrap();
        assert!(body.contains("Dune: Part Two (2024)"), "{body}");
    }

    /// §5.3 through §5.7: Profile 5 is a hard fail — blacklisted, grab
    /// failed, gap reopened, and the torrent's files untouched.
    #[tokio::test]
    async fn a_dolby_vision_profile_5_file_hard_fails() {
        let h = harness(DV5_PROBE).await;

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        assert!(
            std::fs::read_dir(&h.library).unwrap().next().is_none(),
            "nothing may reach the library"
        );
        assert!(h.downloads.join("Dune/Dune.mkv").is_file(), "§7.3");

        let (infohash, normalised, reason): (String, String, String) =
            sqlx::query_as("SELECT infohash, normalised_name, reason FROM blacklist")
                .fetch_one(h.database.pool())
                .await
                .unwrap();
        assert_eq!(infohash, INFOHASH);
        assert_eq!(normalised, arr_parse::normalise(RELEASE_NAME));
        assert_eq!(reason, "dolby_vision_profile");

        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(grab_state, "failed");
        let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(
            movie_state, "missing",
            "the gap reopens for the next candidate"
        );

        assert!(
            h.ntfy_server.received_requests().await.unwrap().is_empty(),
            "§9.5: a hard fail is not notified"
        );
    }

    /// §9.5 *broken*: a full disk notifies the operator once, not every
    /// tick, and re-arms once space frees up.
    #[tokio::test]
    async fn disk_full_notifies_the_operator_once_until_it_clears() {
        let h = harness(HDR10_PROBE).await;
        let error = ImportError::Io {
            action: "hardlink into",
            path: PathBuf::from("/mnt/media/x.mkv"),
            source: io::Error::from(io::ErrorKind::StorageFull),
        };

        h.action.notify_if_disk_full(&error).await;
        h.action.notify_if_disk_full(&error).await;

        let notifications = h.ntfy_server.received_requests().await.unwrap();
        assert_eq!(notifications.len(), 1, "debounced while still full");
        assert_eq!(notifications[0].url.path(), "/operator-topic");

        h.action.clear_disk_full().await;
        h.action.notify_if_disk_full(&error).await;
        assert_eq!(
            h.ntfy_server.received_requests().await.unwrap().len(),
            2,
            "re-arms once space frees up"
        );
    }

    /// §5.7 soft fail: watchable but not what was asked. It imports, and the
    /// row carries the relaxed rule — the file must never read as a clean
    /// match. The torrent is untouched either way (§7.3).
    #[tokio::test]
    async fn an_english_only_kids_import_carries_a_waiver() {
        let h = harness(HDR10_PROBE).await;
        // The kids policy requires Portuguese audio; `allow_english_audio`
        // turns that hard fail into a waiver (§5.2, §5.7).
        let kids_library = h.library.join("kids");
        std::fs::create_dir_all(&kids_library).unwrap();
        sqlx::query("UPDATE roots SET path = ? WHERE kind = 'movie' AND audience = 'kids'")
            .bind(kids_library.to_string_lossy().into_owned())
            .execute(h.database.pool())
            .await
            .unwrap();
        sqlx::query(
            r#"UPDATE movies
               SET root_id   = (SELECT id FROM roots
                                WHERE kind = 'movie' AND audience = 'kids'),
                   overrides = '{"allow_english_audio":true}'
               WHERE id = 1"#,
        )
        .execute(h.database.pool())
        .await
        .unwrap();

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        let waiver: Option<String> = sqlx::query_scalar(
            "SELECT json_extract(waiver, '$.rule') FROM media_files
             WHERE owner_kind = 'movie' AND owner_id = 1",
        )
        .fetch_one(h.database.pool())
        .await
        .unwrap();
        assert_eq!(waiver.as_deref(), Some("required_audio"));

        let blacklisted: i64 = sqlx::query_scalar("SELECT count(*) FROM blacklist")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(blacklisted, 0, "a soft fail blacklists nothing");
        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(grab_state, "imported");
        assert!(
            h.downloads.join("Dune/Dune.mkv").is_file(),
            "§7.3: neither failure mode deletes the torrent"
        );
    }

    /// §8: killed between the hardlink and the bookkeeping, a restart
    /// converges instead of failing on the existing destination.
    #[tokio::test]
    async fn a_restart_after_the_link_converges() {
        let h = harness(HDR10_PROBE).await;
        h.action.tick(&h.database).await.unwrap();

        // The crash: the file is placed, the database never heard.
        sqlx::query("DELETE FROM media_files")
            .execute(h.database.pool())
            .await
            .unwrap();
        sqlx::query("UPDATE grabs SET state = 'downloaded', imported_at = NULL")
            .execute(h.database.pool())
            .await
            .unwrap();
        sqlx::query("UPDATE movies SET state = 'downloading'")
            .execute(h.database.pool())
            .await
            .unwrap();

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        let files: i64 = sqlx::query_scalar("SELECT count(*) FROM media_files")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(files, 1);
        let metadata = std::fs::metadata(h.downloads.join("Dune/Dune.mkv")).unwrap();
        assert_eq!(metadata.nlink(), 2, "no second link, no copy");
    }

    /// A settled tick is idle — an imported grab is not a gap.
    #[tokio::test]
    async fn a_second_tick_imports_nothing_new() {
        let h = harness(HDR10_PROBE).await;
        h.action.tick(&h.database).await.unwrap();

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert!(outcomes.is_empty());
    }

    /// #108, overriding §86: a `downloaded` grab whose torrent Transmission
    /// no longer reports — removed by hand, not a policy failure — is marked
    /// `vanished` and parks the movie (`wanted` cleared) instead of
    /// reopening it as a gap, without touching the blacklist.
    #[tokio::test]
    async fn a_vanished_downloaded_grab_parks_the_movie() {
        let dir = tempfile::tempdir().unwrap();
        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
        database.migrate().await.unwrap();
        sqlx::query(
            "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, state)
             SELECT 693134, 'Dune: Part Two', 2024, 'en', id, 'downloading'
             FROM roots WHERE kind = 'movie' AND audience = 'main'",
        )
        .execute(database.pool())
        .await
        .unwrap();
        let release_id = sqlx::query(
            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
             VALUES (7, 'good', ?, 23622320128, 'magnet:x', '{}', 'eligible')",
        )
        .bind(RELEASE_NAME)
        .execute(database.pool())
        .await
        .unwrap()
        .last_insert_rowid();
        sqlx::query(
            "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
             VALUES (?, 'movie', 1, ?, 'downloaded')",
        )
        .bind(release_id)
        .bind(INFOHASH)
        .execute(database.pool())
        .await
        .unwrap();

        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "result": "success",
                "arguments": {"torrents": []}
            })))
            .mount(&server)
            .await;
        let jellyfin_server = start_jellyfin_server().await;
        let ntfy_server = start_ntfy_server().await;
        let action = ImportAction::new(
            TransmissionClient::new(&server.uri()).unwrap(),
            Prober::new().with_binary(fake_ffprobe(dir.path(), HDR10_PROBE)),
            JellyfinClient::new(jellyfin_server.uri(), None).unwrap(),
            Notifier::new(ntfy_server.uri()).unwrap(),
            Some("operator-topic".to_string()),
        );

        let outcomes = action.tick(&database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs WHERE id = 1")
            .fetch_one(database.pool())
            .await
            .unwrap();
        assert_eq!(grab_state, "vanished");
        let (movie_state, wanted): (String, bool) =
            sqlx::query_as("SELECT state, wanted FROM movies WHERE id = 1")
                .fetch_one(database.pool())
                .await
                .unwrap();
        assert_eq!(movie_state, "parked");
        assert!(!wanted, "the leaf intent is cleared, DESIGN.md §4.1");
        let blacklisted: i64 = sqlx::query_scalar("SELECT count(*) FROM blacklist")
            .fetch_one(database.pool())
            .await
            .unwrap();
        assert_eq!(blacklisted, 0, "a vanished torrent is not a policy failure");
    }

    /// Torrent-declared names are untrusted: absolute and `..`-carrying
    /// entries are skipped, and the import proceeds from what remains.
    #[tokio::test]
    async fn hostile_torrent_paths_never_leave_the_download_root() {
        let h = harness_with(
            HDR10_PROBE,
            json!([
                {"name": "../outside.mkv", "length": 13, "bytesCompleted": 13},
                {"name": "/tmp/absolute.mkv", "length": 13, "bytesCompleted": 13},
                {"name": "Dune/Dune.mkv", "length": 13, "bytesCompleted": 13}
            ]),
        )
        .await;

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        assert!(expected_library_file(&h.library).is_file());
        let files: i64 = sqlx::query_scalar("SELECT count(*) FROM media_files")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(files, 1, "only the safe path is imported");
    }

    /// A torrent whose every entry escapes the root has nothing importable:
    /// hard fail, not an escape.
    #[tokio::test]
    async fn a_torrent_of_only_hostile_paths_hard_fails() {
        let h = harness_with(
            HDR10_PROBE,
            json!([{"name": "../../etc/passwd", "length": 13, "bytesCompleted": 13}]),
        )
        .await;

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        let reason: String = sqlx::query_scalar("SELECT reason FROM blacklist")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(reason, "no readable video file");
        assert!(std::fs::read_dir(&h.library).unwrap().next().is_none());
    }

    #[test]
    fn safe_join_refuses_escapes_and_keeps_normal_paths() {
        let root = Path::new("/downloads");
        assert_eq!(
            safe_join(root, Path::new("Dune/./Dune.mkv")),
            Some(PathBuf::from("/downloads/Dune/Dune.mkv"))
        );
        assert_eq!(safe_join(root, Path::new("../outside.mkv")), None);
        assert_eq!(safe_join(root, Path::new("Dune/../../outside.mkv")), None);
        assert_eq!(safe_join(root, Path::new("/etc/passwd")), None);
        assert_eq!(safe_join(root, Path::new("")), None);
    }

    /// The reconcile lane cancels the action after 25 s while one probe may
    /// take 60 s, so results settled on one tick must survive to the next —
    /// otherwise a large torrent restarts from file one forever.
    #[tokio::test]
    async fn probe_results_are_reused_across_calls() {
        let dir = tempfile::tempdir().unwrap();
        let media = dir.path().join("film.mkv");
        std::fs::write(&media, b"bytes").unwrap();
        let counter = dir.path().join("count");
        let script = dir.path().join("ffprobe");
        std::fs::write(
            &script,
            format!(
                "#!/bin/sh\necho x >> {}\ncat <<'PROBE_EOF'\n{HDR10_PROBE}\nPROBE_EOF\n",
                counter.display()
            ),
        )
        .unwrap();
        let mut permissions = std::fs::metadata(&script).unwrap().permissions();
        permissions.set_mode(0o755);
        std::fs::set_permissions(&script, permissions).unwrap();

        let action = ImportAction::new(
            TransmissionClient::new("http://127.0.0.1:1").unwrap(),
            Prober::new().with_binary(&script),
            JellyfinClient::new("http://127.0.0.1:1", None).unwrap(),
            Notifier::new("http://127.0.0.1:1").unwrap(),
            None,
        );
        let paths = vec![media];

        action.probe_all(&paths).await.unwrap();
        action.probe_all(&paths).await.unwrap();
        assert_eq!(
            std::fs::read_to_string(&counter).unwrap().lines().count(),
            1,
            "the second call reuses the first call's result"
        );

        action.forget_probes(&paths).await;
        action.probe_all(&paths).await.unwrap();
        assert_eq!(
            std::fs::read_to_string(&counter).unwrap().lines().count(),
            2,
            "a settled grab's entries are dropped"
        );
    }

    /// The probe outlives a cancelled tick: the detached task deposits its
    /// result after the caller's future is dropped, and the next tick reads
    /// it instead of restarting the same probe forever.
    #[tokio::test]
    async fn a_cancelled_probe_still_deposits_its_result() {
        let dir = tempfile::tempdir().unwrap();
        let media = dir.path().join("film.mkv");
        std::fs::write(&media, b"bytes").unwrap();
        let counter = dir.path().join("count");
        let script = dir.path().join("ffprobe");
        std::fs::write(
            &script,
            format!(
                "#!/bin/sh\nsleep 0.3\necho x >> {}\ncat <<'PROBE_EOF'\n{HDR10_PROBE}\nPROBE_EOF\n",
                counter.display()
            ),
        )
        .unwrap();
        let mut permissions = std::fs::metadata(&script).unwrap().permissions();
        permissions.set_mode(0o755);
        std::fs::set_permissions(&script, permissions).unwrap();
        let action = ImportAction::new(
            TransmissionClient::new("http://127.0.0.1:1").unwrap(),
            Prober::new().with_binary(&script),
            JellyfinClient::new("http://127.0.0.1:1", None).unwrap(),
            Notifier::new("http://127.0.0.1:1").unwrap(),
            None,
        );
        let paths = vec![media];

        // The lane's budget, in miniature: the tick is cancelled mid-probe.
        let cancelled = tokio::time::timeout(
            std::time::Duration::from_millis(50),
            action.probe_all(&paths),
        )
        .await;
        assert!(cancelled.is_err());

        // The detached probe finishes on its own and deposits the result.
        tokio::time::sleep(std::time::Duration::from_millis(600)).await;
        let files = action.probe_all(&paths).await.unwrap();
        assert_eq!(files.len(), 1);
        assert_eq!(
            std::fs::read_to_string(&counter).unwrap().lines().count(),
            1,
            "the next tick reused the deposited result instead of re-probing"
        );
    }

    /// The TV probe: a 2160p HDR10 file with an English track, sized inside
    /// the 2160p band.
    const TV_HDR10_PROBE: &str = r#"{
        "format": {"format_name": "matroska,webm", "duration": "3300.0", "size": "10737418240"},
        "streams": [
            {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600,
             "color_transfer": "smpte2084"},
            {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}}
        ]
    }"#;

    const TV_DV5_PROBE: &str = r#"{
        "format": {"format_name": "matroska,webm", "duration": "3300.0", "size": "10737418240"},
        "streams": [
            {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600,
             "color_transfer": "smpte2084",
             "side_data_list": [{"side_data_type": "DOVI configuration record", "dv_profile": 5}]},
            {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}}
        ]
    }"#;

    const PACK_RELEASE_NAME: &str = "Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos";

    struct TvHarness {
        _dir: tempfile::TempDir,
        database: Db,
        downloads: PathBuf,
        library: PathBuf,
        action: ImportAction,
        _server: MockServer,
        ntfy_server: MockServer,
    }

    /// A downloaded season-pack grab for Fallout S01E01-E02, its two files
    /// sitting in the download root.
    async fn tv_harness(media_json: &str) -> TvHarness {
        let dir = tempfile::tempdir().unwrap();
        let downloads = dir.path().join("downloads");
        let library = dir.path().join("library");
        std::fs::create_dir_all(downloads.join("Fallout.S01")).unwrap();
        std::fs::create_dir_all(&library).unwrap();
        std::fs::write(downloads.join("Fallout.S01/Fallout.S01E01.mkv"), b"e1").unwrap();
        std::fs::write(downloads.join("Fallout.S01/Fallout.S01E02.mkv"), b"e2").unwrap();

        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
        database.migrate().await.unwrap();
        let library_text = library.to_string_lossy().into_owned();
        sqlx::query("UPDATE roots SET path = ? WHERE kind = 'tv' AND audience = 'main'")
            .bind(&library_text)
            .execute(database.pool())
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO series (tmdb_id, title, year, original_language, root_id)
             SELECT 106379, 'Fallout', 2024, 'en', id
             FROM roots WHERE kind = 'tv' AND audience = 'main'",
        )
        .execute(database.pool())
        .await
        .unwrap();
        let season_id: i64 = sqlx::query_scalar(
            "INSERT INTO seasons (series_id, number) VALUES (1, 1) RETURNING id",
        )
        .fetch_one(database.pool())
        .await
        .unwrap();
        for number in 1..=2 {
            sqlx::query(
                "INSERT INTO episodes (season_id, number, title, air_date, wanted, state)
                 VALUES (?, ?, ?, '2024-04-11', 1, 'downloading')",
            )
            .bind(season_id)
            .bind(number)
            .bind(format!("The Episode {number}"))
            .execute(database.pool())
            .await
            .unwrap();
        }
        let release_id = sqlx::query(
            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
             VALUES (7, 'pack', ?, 85899345920, 'magnet:x', '{}', 'eligible')",
        )
        .bind(PACK_RELEASE_NAME)
        .execute(database.pool())
        .await
        .unwrap()
        .last_insert_rowid();
        sqlx::query(
            "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
             VALUES (?, 'season', ?, ?, 'downloaded')",
        )
        .bind(release_id)
        .bind(season_id)
        .bind(INFOHASH)
        .execute(database.pool())
        .await
        .unwrap();
        insert_owner(&database, "series", 1, "Bob", "bob-topic").await;

        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "result": "success",
                "arguments": {"torrents": [{
                    "hashString": INFOHASH,
                    "downloadDir": downloads.to_string_lossy(),
                    "files": [
                        {"name": "Fallout.S01/Fallout.S01E01.mkv", "length": 2, "bytesCompleted": 2},
                        {"name": "Fallout.S01/Fallout.S01E02.mkv", "length": 2, "bytesCompleted": 2}
                    ]
                }]}
            })))
            .mount(&server)
            .await;
        let jellyfin_server = start_jellyfin_server().await;
        let ntfy_server = start_ntfy_server().await;

        let prober = Prober::new().with_binary(fake_ffprobe(dir.path(), media_json));
        let jellyfin = JellyfinClient::new(jellyfin_server.uri(), None).unwrap();
        let notifier = Notifier::new(ntfy_server.uri()).unwrap();
        let action = ImportAction::new(
            TransmissionClient::new(&server.uri()).unwrap(),
            prober,
            jellyfin,
            notifier,
            Some("operator-topic".to_string()),
        );

        TvHarness {
            _dir: dir,
            database,
            downloads,
            library,
            action,
            _server: server,
            ntfy_server,
        }
    }

    fn expected_episode_file(library: &Path, number: u16) -> PathBuf {
        library
            .join("Fallout (2024) [tmdbid-106379]")
            .join("Season 01")
            .join(format!(
                "Fallout (2024) - S01E{number:02} - The Episode {number} [2160p][WEB-DL][HDR10].mkv"
            ))
    }

    /// A season pack lands each episode file on the §7.4 TV layout.
    #[tokio::test]
    async fn a_season_pack_imports_every_episode() {
        let h = tv_harness(TV_HDR10_PROBE).await;

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        for number in 1..=2u16 {
            let file = expected_episode_file(&h.library, number);
            assert!(file.is_file(), "missing {}", file.display());
        }
        let states: Vec<String> = sqlx::query_scalar("SELECT state FROM episodes ORDER BY number")
            .fetch_all(h.database.pool())
            .await
            .unwrap();
        assert_eq!(states, vec!["available", "available"]);
        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(grab_state, "imported");
        assert!(
            h.downloads.join("Fallout.S01/Fallout.S01E01.mkv").is_file(),
            "§7.3: the torrent keeps seeding"
        );

        let notifications = h.ntfy_server.received_requests().await.unwrap();
        assert_eq!(
            notifications.len(),
            1,
            "one notification per grab, not per episode"
        );
        assert_eq!(notifications[0].url.path(), "/bob-topic");
        let body = String::from_utf8(notifications[0].body.clone()).unwrap();
        assert!(body.contains("Fallout (2024)"), "{body}");
    }

    /// The fourth acceptance case: a pack containing an episode already on
    /// disk must not re-import what exists.
    #[tokio::test]
    async fn a_season_pack_never_reimports_an_episode_already_on_disk() {
        let h = tv_harness(TV_HDR10_PROBE).await;
        let existing = h.library.join("existing-e01.mkv");
        std::fs::write(&existing, b"the copy that is already there").unwrap();
        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size)
             SELECT 'episode', id, ?, 30 FROM episodes WHERE number = 1",
        )
        .bind(existing.to_string_lossy().into_owned())
        .execute(h.database.pool())
        .await
        .unwrap();
        sqlx::query("UPDATE episodes SET state = 'available' WHERE number = 1")
            .execute(h.database.pool())
            .await
            .unwrap();

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        assert!(
            !expected_episode_file(&h.library, 1).exists(),
            "episode 1 is on disk already and must not be re-imported"
        );
        assert!(expected_episode_file(&h.library, 2).is_file());
        let episode_one_files: Vec<(String, i64)> = sqlx::query_as(
            "SELECT f.path, f.size FROM media_files f
             JOIN episodes e ON e.id = f.owner_id
             WHERE f.owner_kind = 'episode' AND e.number = 1",
        )
        .fetch_all(h.database.pool())
        .await
        .unwrap();
        assert_eq!(
            episode_one_files,
            vec![(existing.to_string_lossy().into_owned(), 30)],
            "episode 1 keeps exactly its pre-existing file row"
        );
        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(grab_state, "imported");
    }

    /// The third acceptance case, import side: a pack whose file hard-fails
    /// blacklists that release and reopens the episodes — it does not
    /// blacklist or block the season.
    #[tokio::test]
    async fn a_hard_failed_pack_reopens_the_season_per_episode() {
        let h = tv_harness(TV_DV5_PROBE).await;

        let outcomes = h.action.tick(&h.database).await.unwrap();

        assert_eq!(outcomes.len(), 1);
        assert!(
            std::fs::read_dir(&h.library).unwrap().next().is_none(),
            "nothing may reach the library"
        );
        let blacklist: Vec<(String, String)> =
            sqlx::query_as("SELECT normalised_name, reason FROM blacklist")
                .fetch_all(h.database.pool())
                .await
                .unwrap();
        assert_eq!(
            blacklist,
            vec![(
                arr_parse::normalise(PACK_RELEASE_NAME),
                "dolby_vision_profile".to_owned()
            )],
            "only the release is blacklisted, never the season"
        );
        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
            .fetch_one(h.database.pool())
            .await
            .unwrap();
        assert_eq!(grab_state, "failed");
        let states: Vec<(String, bool)> =
            sqlx::query_as("SELECT state, wanted FROM episodes ORDER BY number")
                .fetch_all(h.database.pool())
                .await
                .unwrap();
        assert_eq!(
            states,
            vec![("missing".to_owned(), true), ("missing".to_owned(), true)],
            "the gap reopens per episode, still wanted"
        );
        assert!(
            h.downloads.join("Fallout.S01/Fallout.S01E01.mkv").is_file(),
            "§7.3: the torrent is untouched"
        );
    }

    /// The `EXDEV` fallback path lands whole files via rename (§7.2).
    #[test]
    fn the_copy_fallback_lands_a_whole_file_and_cleans_up() {
        let dir = tempfile::tempdir().unwrap();
        let source = dir.path().join("source.mkv");
        std::fs::write(&source, b"feature bytes").unwrap();
        let destination = dir.path().join("library").join("feature.mkv");
        std::fs::create_dir_all(destination.parent().unwrap()).unwrap();

        let placement = copy_into_place(&source, &destination).unwrap();

        assert_eq!(placement, Placement::Copied);
        assert_eq!(std::fs::read(&destination).unwrap(), b"feature bytes");
        assert!(
            std::fs::read_dir(destination.parent().unwrap())
                .unwrap()
                .all(|entry| !entry
                    .unwrap()
                    .file_name()
                    .to_string_lossy()
                    .contains("partial")),
            "no partial file left behind"
        );
    }
}