use arr_core::policy::Candidate;
use arr_core::score::score;
use arr_core::{ParsedRelease, Policy};
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::Json;
use serde::{Deserialize, Serialize};
use utoipa::{IntoParams, ToSchema};

use crate::owners::Owner;
use crate::state::{AppState, MovieCommand};

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct Movie {
    pub id: i64,
    pub tmdb_id: i64,
    pub title: String,
    pub year: Option<i64>,
    pub original_language: Option<String>,
    pub root_id: i64,
    pub wanted: bool,
    pub overrides: serde_json::Value,
    pub state: String,
    pub blocked: bool,
    pub search_attempts: i64,
    pub last_searched_at: Option<String>,
    /// The relaxed rule recorded when a file was imported under a waiver
    /// (§5.7), so the UI can show "English, no dub" instead of a clean match.
    pub waiver: Option<serde_json::Value>,
}

#[derive(Debug, Deserialize, ToSchema)]
pub struct CreateMovie {
    pub tmdb_id: i64,
    pub title: String,
    pub year: Option<i64>,
    pub original_language: Option<String>,
    pub root_id: i64,
    #[serde(default = "default_true")]
    pub wanted: bool,
    #[serde(default)]
    pub blocked: bool,
    #[serde(default = "empty_overrides")]
    pub overrides: serde_json::Value,
}

#[derive(Debug, Deserialize, ToSchema)]
pub struct UpdateMovie {
    pub title: Option<String>,
    pub year: Option<Option<i64>>,
    pub original_language: Option<Option<String>>,
    pub root_id: Option<i64>,
    pub wanted: Option<bool>,
    pub blocked: Option<bool>,
    pub overrides: Option<serde_json::Value>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct Release {
    pub id: i64,
    pub indexer_id: i64,
    pub guid: String,
    pub name: String,
    pub size: i64,
    pub seeders: Option<i64>,
    pub publish_date: Option<String>,
    pub download_url: String,
    pub parsed: serde_json::Value,
    pub score: Option<f64>,
    pub verdict: Option<String>,
    pub rejected_rule: Option<String>,
}

/// A library file and what it cost to accept it (`DESIGN.md` §5.7).
///
/// `waiver` names the rule that was relaxed to let a soft fail in. It is the
/// difference between a file that satisfies the policy and one that merely
/// plays, so it travels with the file everywhere the file does.
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct MovieFile {
    pub id: i64,
    pub path: String,
    pub size: i64,
    /// What `ffprobe` found (§5.6).
    pub probed: Option<serde_json::Value>,
    /// The relaxed rule's name, or `null` for a clean import. Shares its
    /// vocabulary with `Release::rejected_rule`.
    pub waiver: Option<String>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct AttentionQueues {
    pub no_pt_source: Vec<Movie>,
    pub needs_decision: Vec<Movie>,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct Accepted {
    pub accepted: bool,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ErrorBody {
    pub error: String,
}

#[derive(Debug)]
pub enum ApiError {
    NotFound,
    SeriesNotFound,
    SeasonNotFound,
    EpisodeNotFound,
    OwnerNotFound,
    Conflict(String),
    Invalid(String),
    Unavailable,
    Database(String),
    /// A library delete that could not touch the disk. Named separately from
    /// [`Self::Database`] because the row is still there and a retry is the
    /// right next move.
    Filesystem(String),
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, error) = match self {
            Self::NotFound => (StatusCode::NOT_FOUND, "movie not found".to_string()),
            Self::SeriesNotFound => (StatusCode::NOT_FOUND, "series not found".to_string()),
            Self::SeasonNotFound => (StatusCode::NOT_FOUND, "season not found".to_string()),
            Self::EpisodeNotFound => (StatusCode::NOT_FOUND, "episode not found".to_string()),
            Self::OwnerNotFound => (StatusCode::NOT_FOUND, "owner not found".to_string()),
            Self::Conflict(error) => (StatusCode::CONFLICT, error),
            Self::Invalid(error) => (StatusCode::UNPROCESSABLE_ENTITY, error),
            Self::Unavailable => (
                StatusCode::SERVICE_UNAVAILABLE,
                "database unavailable".into(),
            ),
            Self::Database(error) => {
                tracing::error!(%error, "API database error");
                (StatusCode::INTERNAL_SERVER_ERROR, "database error".into())
            }
            Self::Filesystem(error) => {
                tracing::error!(%error, "API filesystem error");
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("files not removed: {error}"),
                )
            }
        };
        (status, Json(ErrorBody { error })).into_response()
    }
}

impl From<sqlx::Error> for ApiError {
    fn from(error: sqlx::Error) -> Self {
        if matches!(error, sqlx::Error::RowNotFound) {
            Self::NotFound
        } else if let sqlx::Error::Database(database) = &error {
            if database.is_unique_violation() || database.is_foreign_key_violation() {
                return Self::Conflict(database.message().to_string());
            }
            Self::Database(error.to_string())
        } else {
            Self::Database(error.to_string())
        }
    }
}

fn default_true() -> bool {
    true
}

fn empty_overrides() -> serde_json::Value {
    serde_json::json!({})
}

fn validate_overrides(value: &serde_json::Value) -> Result<(), ApiError> {
    let Some(object) = value.as_object() else {
        return Err(ApiError::Invalid("overrides must be an object".into()));
    };
    if object
        .keys()
        .any(|key| key != "only_4k" && key != "allow_english_audio")
    {
        return Err(ApiError::Invalid(
            "overrides supports only only_4k and allow_english_audio".into(),
        ));
    }
    if object.values().any(|value| !value.is_boolean()) {
        return Err(ApiError::Invalid("override values must be booleans".into()));
    }
    Ok(())
}

pub(crate) fn pool(state: &AppState) -> Result<&sqlx::SqlitePool, ApiError> {
    state
        .database()
        .map(arr_db::Db::pool)
        .ok_or(ApiError::Unavailable)
}

async fn load_movie(state: &AppState, id: i64) -> Result<Movie, ApiError> {
    Ok(sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies WHERE id = ?"#, id)
        .fetch_one(pool(state)?)
        .await?)
}

#[derive(Debug, Deserialize, IntoParams)]
pub struct ListMoviesQuery {
    /// Restrict to movies tagged with this owner (DESIGN.md §4.3).
    pub owner_id: Option<i64>,
}

#[utoipa::path(
    get, path = "/api/movies", tag = "movies",
    params(ListMoviesQuery),
    responses(
        (status = 200, body = [Movie]),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn list(
    State(state): State<AppState>,
    Query(query): Query<ListMoviesQuery>,
) -> Result<Json<Vec<Movie>>, ApiError> {
    let movies = if let Some(owner_id) = query.owner_id {
        sqlx::query_as!(Movie, r#"SELECT m.id AS "id!: i64", m.tmdb_id AS "tmdb_id!: i64", m.title AS "title!: String", m.year, m.original_language, m.root_id AS "root_id!: i64", m.wanted AS "wanted!: bool", m.overrides AS "overrides!: serde_json::Value", m.state AS "state!: String", m.blocked AS "blocked!: bool", m.search_attempts AS "search_attempts!: i64", m.last_searched_at, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = m.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies m JOIN title_owners t ON t.title_kind = 'movie' AND t.title_id = m.id WHERE t.owner_id = ? ORDER BY m.title, m.year, m.id"#, owner_id)
            .fetch_all(pool(&state)?)
            .await?
    } else {
        sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies ORDER BY title, year, id"#)
            .fetch_all(pool(&state)?)
            .await?
    };
    Ok(Json(movies))
}

#[utoipa::path(
    post, path = "/api/movies", tag = "movies", request_body = CreateMovie,
    responses(
        (status = 201, body = Movie),
        (status = 409, body = ErrorBody),
        (status = 422, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn create(
    State(state): State<AppState>,
    Json(input): Json<CreateMovie>,
) -> Result<(StatusCode, Json<Movie>), ApiError> {
    if input.title.trim().is_empty() || input.tmdb_id <= 0 {
        return Err(ApiError::Invalid("tmdb_id and title are required".into()));
    }
    validate_overrides(&input.overrides)?;
    let movie_root = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM roots WHERE id = ? AND kind = 'movie') AS 'exists!: bool'",
        input.root_id
    )
    .fetch_one(pool(&state)?)
    .await?;
    if !movie_root {
        return Err(ApiError::Invalid("root_id must name a movie root".into()));
    }
    let overrides = serde_json::to_string(&input.overrides)
        .map_err(|error| ApiError::Invalid(error.to_string()))?;
    let title = input.title.trim();
    let result = sqlx::query!(
        "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, wanted, blocked, overrides) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        input.tmdb_id, title, input.year, input.original_language, input.root_id,
        input.wanted, input.blocked, overrides
    )
    .execute(pool(&state)?)
    .await?;
    Ok((
        StatusCode::CREATED,
        Json(load_movie(&state, result.last_insert_rowid()).await?),
    ))
}

#[utoipa::path(
    get, path = "/api/movies/{movie_id}", tag = "movies",
    params(("movie_id" = i64, Path, description = "Movie row id")),
    responses(
        (status = 200, body = Movie),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn get(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Movie>, ApiError> {
    Ok(Json(load_movie(&state, id).await?))
}

#[utoipa::path(
    patch, path = "/api/movies/{movie_id}", tag = "movies", request_body = UpdateMovie,
    params(("movie_id" = i64, Path, description = "Movie row id")),
    responses(
        (status = 200, body = Movie),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 422, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn update(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    Json(input): Json<UpdateMovie>,
) -> Result<Json<Movie>, ApiError> {
    let current = load_movie(&state, id).await?;
    let title = input.title.unwrap_or(current.title);
    if title.trim().is_empty() {
        return Err(ApiError::Invalid("title cannot be empty".into()));
    }
    let overrides = input.overrides.unwrap_or(current.overrides);
    validate_overrides(&overrides)?;
    let overrides =
        serde_json::to_string(&overrides).map_err(|error| ApiError::Invalid(error.to_string()))?;
    let title = title.trim();
    let year = input.year.unwrap_or(current.year);
    let original_language = input.original_language.unwrap_or(current.original_language);
    let root_id = input.root_id.unwrap_or(current.root_id);
    if input.root_id.is_some() {
        let movie_root = sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM roots WHERE id = ? AND kind = 'movie') AS 'exists!: bool'",
            root_id
        )
        .fetch_one(pool(&state)?)
        .await?;
        if !movie_root {
            return Err(ApiError::Invalid("root_id must name a movie root".into()));
        }
    }
    let wanted = input.wanted.unwrap_or(current.wanted);
    let blocked = input.blocked.unwrap_or(current.blocked);
    sqlx::query!("UPDATE movies SET title = ?, year = ?, original_language = ?, root_id = ?, wanted = ?, blocked = ?, overrides = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?", title, year, original_language, root_id, wanted, blocked, overrides, id)
        .execute(pool(&state)?)
        .await?;
    Ok(Json(load_movie(&state, id).await?))
}

#[utoipa::path(
    delete, path = "/api/movies/{movie_id}", tag = "movies",
    params(("movie_id" = i64, Path, description = "Movie row id")),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn delete(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<StatusCode, ApiError> {
    // The row is loaded first so a missing movie is 404 before anything
    // touches the disk.
    load_movie(&state, id).await?;
    remove_library_files(&state, id).await?;
    // `media_files.path` is UNIQUE and the owner is polymorphic, so nothing
    // cascades: leaving the rows behind would block re-importing the same
    // path after a re-add. Owner tags go with the title they tagged.
    sqlx::query!(
        "DELETE FROM media_files WHERE owner_kind = 'movie' AND owner_id = ?",
        id
    )
    .execute(pool(&state)?)
    .await?;
    sqlx::query!(
        "DELETE FROM title_owners WHERE title_kind = 'movie' AND title_id = ?",
        id
    )
    .execute(pool(&state)?)
    .await?;
    let result = sqlx::query!("DELETE FROM movies WHERE id = ?", id)
        .execute(pool(&state)?)
        .await?;
    if result.rows_affected() == 0 {
        return Err(ApiError::NotFound);
    }
    Ok(StatusCode::NO_CONTENT)
}

/// Unlink everything this title put under its root.
///
/// The service knows only what it wrote (§2), so the targets come from
/// `media_files`, never from a scan and never from re-deriving the §7.4 name
/// — a title renamed after import would derive a folder that does not exist
/// while the real one stayed. Deleting the folder rather than the file is
/// what makes the delete atomic (§7.4): sidecar subtitles and artwork go
/// with it.
///
/// The torrent is untouched (§7.3). It keeps seeding under its own rule and
/// the reaper deletes it; a hardlinked file loses only its library name.
///
/// Failure leaves the database alone, so the operator sees the title still
/// there and can retry rather than losing the record of what is on disk.
async fn remove_library_files(state: &AppState, id: i64) -> Result<(), ApiError> {
    let root = sqlx::query_scalar!(
        r#"SELECT r.path AS "path!: String" FROM roots r JOIN movies m ON m.root_id = r.id WHERE m.id = ?"#,
        id
    )
    .fetch_one(pool(state)?)
    .await?;
    let paths = sqlx::query_scalar!(
        r#"SELECT path AS "path!: String" FROM media_files WHERE owner_kind = 'movie' AND owner_id = ?"#,
        id
    )
    .fetch_all(pool(state)?)
    .await?;

    let mut targets: Vec<std::path::PathBuf> = Vec::new();
    for path in &paths {
        let Some(target) = title_target(&root, path) else {
            // Outside its own root: not ours to delete. The row still goes,
            // so the operator sees the title leave and the file stay.
            tracing::warn!(%path, %root, "media file is outside its root, not deleted");
            continue;
        };
        if !targets.contains(&target) {
            targets.push(target);
        }
    }

    for target in targets {
        let metadata = match tokio::fs::symlink_metadata(&target).await {
            Ok(metadata) => metadata,
            // Already gone is the state we wanted.
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
            Err(error) => return Err(ApiError::Filesystem(error.to_string())),
        };
        let removed = if metadata.is_dir() {
            tokio::fs::remove_dir_all(&target).await
        } else {
            tokio::fs::remove_file(&target).await
        };
        match removed {
            Ok(()) => tracing::info!(target = %target.display(), "removed library files"),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(ApiError::Filesystem(error.to_string())),
        }
    }
    Ok(())
}

/// What to unlink for one library file: the title folder directly under the
/// root, or the file itself when it sits in the root with no folder of its
/// own. `None` when the file is not under the root at all, which is the
/// guard that keeps a delete inside the library it belongs to.
fn title_target(root: &str, file: &str) -> Option<std::path::PathBuf> {
    let root = std::path::Path::new(root);
    let relative = std::path::Path::new(file).strip_prefix(root).ok()?;
    let first = relative.components().next()?;
    let std::path::Component::Normal(name) = first else {
        // `..` or a root component would climb out of the library.
        return None;
    };
    Some(root.join(name))
}

#[utoipa::path(
    post, path = "/api/movies/{movie_id}/search", tag = "movies",
    params(("movie_id" = i64, Path, description = "Movie row id")),
    responses(
        (status = 202, body = Accepted),
        (status = 404, body = ErrorBody),
        (status = 409, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn search(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<(StatusCode, Json<Accepted>), ApiError> {
    let movie = load_movie(&state, id).await?;
    if movie.blocked {
        return Err(ApiError::Conflict("movie is blocked".into()));
    }
    state
        .send_movie_command(MovieCommand::Search { movie_id: id })
        .map_err(|_| ApiError::Unavailable)?;
    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}

#[utoipa::path(
    get, path = "/api/movies/{movie_id}/releases", tag = "movies",
    params(("movie_id" = i64, Path, description = "Movie row id")),
    responses(
        (status = 200, body = [Release]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn releases(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<Release>>, ApiError> {
    load_movie(&state, id).await?;
    let mut releases = sqlx::query_as!(Release, r#"SELECT r.id AS "id!: i64", r.indexer_id AS "indexer_id!: i64", r.guid AS "guid!: String", r.name AS "name!: String", r.size AS "size!: i64", r.seeders, r.publish_date, r.download_url AS "download_url!: String", r.parsed AS "parsed!: serde_json::Value", r.score, r.verdict, r.rejected_rule FROM releases r JOIN movie_releases mr ON mr.release_id = r.id WHERE mr.movie_id = ? ORDER BY CASE r.verdict WHEN 'eligible' THEN 0 WHEN 'waived' THEN 1 ELSE 2 END, r.score DESC, r.id"#, id)
    .fetch_all(pool(&state)?)
    .await?;
    let policy = state
        .database()
        .ok_or(ApiError::Unavailable)?
        .movie_policy(id)
        .await
        .map_err(|error| ApiError::Database(error.to_string()))?
        .ok_or(ApiError::NotFound)?
        .policy;
    rescore(&mut releases, &policy)?;
    Ok(Json(releases))
}

/// Recompute each release's score against the current policy and re-sort
/// (`DESIGN.md` §5.5). The stored `score` column is grab-time truth for the
/// daemon's winner pick; the deck the user sees must reflect policy tuning
/// made since the release was last searched, not whatever was true then.
///
/// Score magnitudes stay far below `f64`'s 52-bit mantissa (they are sums of
/// policy weights in the thousands), so the `i64` -> `f64` cast into the
/// column's storage type is exact.
#[allow(clippy::cast_precision_loss)]
pub(crate) fn rescore(releases: &mut [Release], policy: &Policy) -> Result<(), ApiError> {
    let mut totals = Vec::with_capacity(releases.len());
    for release in releases.iter() {
        let parsed: ParsedRelease = serde_json::from_value(release.parsed.clone())
            .map_err(|error| ApiError::Database(error.to_string()))?;
        let seeders = u32::try_from(release.seeders.unwrap_or(0)).unwrap_or(u32::MAX);
        let size = u64::try_from(release.size).unwrap_or(0);
        totals.push(score(policy, Candidate::PreGrab(&parsed), size, seeders).total);
    }
    let mut indices: Vec<usize> = (0..releases.len()).collect();
    indices.sort_by_key(|&i| (bucket(releases[i].verdict.as_deref()), -totals[i]));

    for (release, &total) in releases.iter_mut().zip(&totals) {
        release.score = Some(total as f64);
    }
    let sorted: Vec<Release> = indices.into_iter().map(|i| releases[i].clone()).collect();
    releases.clone_from_slice(&sorted);
    Ok(())
}

fn bucket(verdict: Option<&str>) -> u8 {
    match verdict {
        Some("eligible") => 0,
        Some("waived") => 1,
        _ => 2,
    }
}

#[utoipa::path(
    get, path = "/api/movies/{movie_id}/files", tag = "movies",
    params(("movie_id" = i64, Path, description = "Movie row id")),
    responses(
        (status = 200, body = [MovieFile]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn files(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<MovieFile>>, ApiError> {
    load_movie(&state, id).await?;
    let files = sqlx::query_as!(MovieFile, r#"SELECT id AS "id!: i64", path AS "path!: String", size AS "size!: i64", probed AS "probed?: serde_json::Value", json_extract(waiver, '$.rule') AS "waiver?: String" FROM media_files WHERE owner_kind = 'movie' AND owner_id = ? ORDER BY path"#, id)
    .fetch_all(pool(&state)?)
    .await?;
    Ok(Json(files))
}

#[utoipa::path(
    post, path = "/api/movies/{movie_id}/releases/{release_id}/grab", tag = "movies",
    params(("movie_id" = i64, Path), ("release_id" = i64, Path)),
    responses(
        (status = 202, body = Accepted),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn grab(
    State(state): State<AppState>,
    Path((movie_id, release_id)): Path<(i64, i64)>,
) -> Result<(StatusCode, Json<Accepted>), ApiError> {
    let exists = sqlx::query_scalar!("SELECT EXISTS(SELECT 1 FROM movie_releases WHERE movie_id = ? AND release_id = ?) AS 'exists!: bool'", movie_id, release_id)
    .fetch_one(pool(&state)?)
    .await?;
    if !exists {
        return Err(ApiError::NotFound);
    }
    state
        .send_movie_command(MovieCommand::Grab {
            movie_id,
            release_id,
        })
        .map_err(|_| ApiError::Unavailable)?;
    Ok((StatusCode::ACCEPTED, Json(Accepted { accepted: true })))
}

#[utoipa::path(
    get, path = "/api/queues/attention", tag = "movies",
    responses(
        (status = 200, body = AttentionQueues),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQueues>, ApiError> {
    let no_pt_source = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies WHERE id IN (SELECT m.id FROM movies m JOIN roots root ON root.id = m.root_id WHERE root.audience = 'kids' AND m.wanted = 1 AND m.blocked = 0 AND m.state = 'missing' AND m.search_attempts > 0 AND NOT EXISTS (SELECT 1 FROM movie_releases mr JOIN releases r ON r.id = mr.release_id WHERE mr.movie_id = m.id AND r.verdict IN ('eligible', 'waived'))) ORDER BY title"#)
        .fetch_all(pool(&state)?)
        .await?;
    let needs_decision = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at, (SELECT f.waiver FROM media_files f WHERE f.owner_kind = 'movie' AND f.owner_id = movies.id AND f.waiver IS NOT NULL ORDER BY f.id LIMIT 1) AS "waiver?: serde_json::Value" FROM movies WHERE (SELECT count(DISTINCT g.release_id) FROM grabs g WHERE g.target_kind = 'movie' AND g.target_id = movies.id AND g.state = 'failed') >= 2 ORDER BY title"#)
        .fetch_all(pool(&state)?)
        .await?;
    Ok(Json(AttentionQueues {
        no_pt_source,
        needs_decision,
    }))
}

#[utoipa::path(
    get, path = "/api/movies/{movie_id}/owners", tag = "movies",
    params(("movie_id" = i64, Path, description = "Movie row id")),
    responses(
        (status = 200, body = [Owner]),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn list_owners(
    State(state): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<Owner>>, ApiError> {
    load_movie(&state, id).await?;
    let owners = sqlx::query_as!(
        Owner,
        r#"SELECT o.id AS "id!: i64", o.name AS "name!: String", o.ntfy_topic AS "ntfy_topic!: String"
           FROM owners o
           JOIN title_owners t ON t.owner_id = o.id
           WHERE t.title_kind = 'movie' AND t.title_id = ?
           ORDER BY o.name"#,
        id
    )
    .fetch_all(pool(&state)?)
    .await?;
    Ok(Json(owners))
}

#[utoipa::path(
    put, path = "/api/movies/{movie_id}/owners/{owner_id}", tag = "movies",
    params(("movie_id" = i64, Path), ("owner_id" = i64, Path)),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn tag_owner(
    State(state): State<AppState>,
    Path((movie_id, owner_id)): Path<(i64, i64)>,
) -> Result<StatusCode, ApiError> {
    load_movie(&state, movie_id).await?;
    let owner_exists = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM owners WHERE id = ?) AS 'exists!: bool'",
        owner_id
    )
    .fetch_one(pool(&state)?)
    .await?;
    if !owner_exists {
        return Err(ApiError::OwnerNotFound);
    }
    sqlx::query!(
        "INSERT INTO title_owners (title_kind, title_id, owner_id) VALUES ('movie', ?, ?) ON CONFLICT DO NOTHING",
        movie_id, owner_id
    )
    .execute(pool(&state)?)
    .await?;
    Ok(StatusCode::NO_CONTENT)
}

#[utoipa::path(
    delete, path = "/api/movies/{movie_id}/owners/{owner_id}", tag = "movies",
    params(("movie_id" = i64, Path), ("owner_id" = i64, Path)),
    responses(
        (status = 204),
        (status = 404, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn untag_owner(
    State(state): State<AppState>,
    Path((movie_id, owner_id)): Path<(i64, i64)>,
) -> Result<StatusCode, ApiError> {
    load_movie(&state, movie_id).await?;
    sqlx::query!(
        "DELETE FROM title_owners WHERE title_kind = 'movie' AND title_id = ? AND owner_id = ?",
        movie_id,
        owner_id
    )
    .execute(pool(&state)?)
    .await?;
    Ok(StatusCode::NO_CONTENT)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{router, Upstreams};

    async fn application() -> (tempfile::TempDir, AppState, String) {
        let dir = tempfile::tempdir().expect("tempdir");
        let database = arr_db::Db::connect(dir.path().join("arr.db"))
            .await
            .expect("connect database");
        database.migrate().await.expect("migrate database");
        let state = AppState::new(Upstreams::new(
            "http://127.0.0.1:1".into(),
            "http://127.0.0.1:1".into(),
        ))
        .expect("state")
        .with_database(database);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let address = listener.local_addr().expect("address");
        let app = router(state.clone());
        tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
        (dir, state, format!("http://{address}"))
    }

    async fn add_movie(base: &str, tmdb_id: i64, root_id: i64) -> serde_json::Value {
        let response = reqwest::Client::new()
            .post(format!("{base}/api/movies"))
            .json(&serde_json::json!({
                "tmdb_id": tmdb_id, "title": "Dune Part Two", "year": 2024,
                "original_language": "en", "root_id": root_id,
                "overrides": {"only_4k": true}
            }))
            .send()
            .await
            .expect("create movie");
        assert_eq!(response.status(), StatusCode::CREATED);
        response.json().await.expect("movie json")
    }

    /// §5.7: a soft-failed import is imported and waived, and the waiver
    /// reaches the API — a file that merely plays must never read as a clean
    /// match.
    #[tokio::test]
    async fn a_movie_file_carries_its_waiver() {
        let (_dir, state, base) = application().await;
        let movie = add_movie(&base, 693_134, 1).await;
        let movie_id = movie["id"].as_i64().expect("movie id");
        let pool = state.database().expect("database").pool();
        sqlx::query(
            r#"INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)
               VALUES ('movie', ?, '/library/dune.mkv', 23622320128,
                       '{"resolution":"1080p"}', '{"rule":"required_audio"}')"#,
        )
        .bind(movie_id)
        .execute(pool)
        .await
        .expect("media file");

        let files: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/movies/{movie_id}/files"))
                .await
                .expect("files")
                .json()
                .await
                .expect("json");

        assert_eq!(files.len(), 1);
        assert_eq!(files[0]["path"], "/library/dune.mkv");
        assert_eq!(files[0]["waiver"], "required_audio");
        assert_eq!(files[0]["probed"]["resolution"], "1080p");
    }

    #[tokio::test]
    async fn crud_preserves_intent_and_overrides() {
        let (_dir, _state, base) = application().await;
        let movie = add_movie(&base, 693_134, 2).await;
        let id = movie["id"].as_i64().expect("id");
        assert_eq!(movie["wanted"], true);
        assert_eq!(movie["overrides"]["only_4k"], true);

        let updated: serde_json::Value = reqwest::Client::new()
            .patch(format!("{base}/api/movies/{id}"))
            .json(&serde_json::json!({
                "wanted": false, "blocked": true,
                "overrides": {"allow_english_audio": true}
            }))
            .send()
            .await
            .expect("update")
            .json()
            .await
            .expect("updated json");
        assert_eq!(updated["wanted"], false);
        assert_eq!(updated["blocked"], true);
        assert_eq!(updated["overrides"]["allow_english_audio"], true);

        let listed: Vec<serde_json::Value> = reqwest::get(format!("{base}/api/movies"))
            .await
            .expect("list")
            .json()
            .await
            .expect("list json");
        assert_eq!(listed.len(), 1);

        let deleted = reqwest::Client::new()
            .delete(format!("{base}/api/movies/{id}"))
            .send()
            .await
            .expect("delete");
        assert_eq!(deleted.status(), StatusCode::NO_CONTENT);
        assert_eq!(
            reqwest::get(format!("{base}/api/movies/{id}"))
                .await
                .expect("get deleted")
                .status(),
            StatusCode::NOT_FOUND
        );
    }

    /// A movie's root is a real directory for the duration of one test, with
    /// the §7.4 folder already in it: one feature and one sidecar subtitle.
    async fn library_on_disk(
        state: &AppState,
        movie_id: i64,
        root: &std::path::Path,
    ) -> std::path::PathBuf {
        let folder = root.join("Dune Part Two (2024) [tmdbid-693134]");
        tokio::fs::create_dir_all(&folder)
            .await
            .expect("create title folder");
        let feature = folder.join("Dune Part Two (2024) [tmdbid-693134] - [2160p].mkv");
        tokio::fs::write(&feature, b"feature").await.expect("write");
        tokio::fs::write(folder.join("dune.pt.srt"), b"subs")
            .await
            .expect("write sidecar");

        let pool = state.database().expect("database").pool();
        let root_path = root.to_str().expect("utf-8 root");
        sqlx::query("UPDATE roots SET path = ? WHERE id = 1")
            .bind(root_path)
            .execute(pool)
            .await
            .expect("point the root at the tempdir");
        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('movie', ?, ?, 7)",
        )
        .bind(movie_id)
        .bind(feature.to_str().expect("utf-8 path"))
        .execute(pool)
        .await
        .expect("media file");
        folder
    }

    /// The §7.4 folder is the unit of deletion, so the sidecar goes with the
    /// feature — and the root itself is never touched.
    #[tokio::test]
    async fn deleting_a_movie_removes_the_whole_title_folder() {
        let (_dir, state, base) = application().await;
        let movie = add_movie(&base, 693_134, 1).await;
        let id = movie["id"].as_i64().expect("id");
        let root = tempfile::tempdir().expect("root");
        let folder = library_on_disk(&state, id, root.path()).await;

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/movies/{id}"))
            .send()
            .await
            .expect("delete");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        assert!(
            !folder.exists(),
            "the title folder and its sidecars are gone"
        );
        assert!(root.path().exists(), "the root survives its titles");

        let orphans: i64 = sqlx::query_scalar(
            "SELECT count(*) FROM media_files WHERE owner_kind = 'movie' AND owner_id = ?",
        )
        .bind(id)
        .fetch_one(state.database().expect("database").pool())
        .await
        .expect("count files");
        assert_eq!(orphans, 0, "the file rows go with the files");
    }

    /// The guard that keeps a delete inside the library: a path that is not
    /// under the title's root is left alone, whatever the row says.
    #[tokio::test]
    async fn a_file_outside_its_root_is_never_unlinked() {
        let (_dir, state, base) = application().await;
        let movie = add_movie(&base, 693_134, 1).await;
        let id = movie["id"].as_i64().expect("id");
        let root = tempfile::tempdir().expect("root");
        let elsewhere = tempfile::tempdir().expect("elsewhere");
        let stray = elsewhere.path().join("not-ours.mkv");
        tokio::fs::write(&stray, b"stray").await.expect("write");

        let pool = state.database().expect("database").pool();
        sqlx::query("UPDATE roots SET path = ? WHERE id = 1")
            .bind(root.path().to_str().expect("utf-8 root"))
            .execute(pool)
            .await
            .expect("point the root at the tempdir");
        sqlx::query(
            "INSERT INTO media_files (owner_kind, owner_id, path, size) VALUES ('movie', ?, ?, 5)",
        )
        .bind(id)
        .bind(stray.to_str().expect("utf-8 path"))
        .execute(pool)
        .await
        .expect("media file");

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/movies/{id}"))
            .send()
            .await
            .expect("delete");
        assert_eq!(response.status(), StatusCode::NO_CONTENT);
        assert!(
            stray.exists(),
            "a path outside the root is not ours to delete"
        );
    }

    #[test]
    fn a_title_target_is_the_folder_directly_under_the_root() {
        let root = "/mnt/media/movies/main";
        assert_eq!(
            title_target(
                root,
                "/mnt/media/movies/main/Dune (2021) [tmdbid-1]/Dune.mkv"
            ),
            Some(std::path::PathBuf::from(
                "/mnt/media/movies/main/Dune (2021) [tmdbid-1]"
            ))
        );
        // A file sitting straight in the root is its own target: deleting the
        // root because a file was misplaced would take the whole library.
        assert_eq!(
            title_target(root, "/mnt/media/movies/main/loose.mkv"),
            Some(std::path::PathBuf::from("/mnt/media/movies/main/loose.mkv"))
        );
        assert_eq!(title_target(root, "/mnt/media/movies/kids/other.mkv"), None);
        assert_eq!(title_target(root, root), None);
    }

    #[tokio::test]
    async fn release_actions_are_scoped_to_the_movie() {
        let (_dir, state, base) = application().await;
        let movie = add_movie(&base, 693_134, 2).await;
        let movie_id = movie["id"].as_i64().expect("id");
        let pool = state.database().expect("database").pool();
        let release_id = sqlx::query("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, score, verdict) VALUES (7, 'guid', 'release', 1000, 'url', '{}', 42, 'eligible')")
            .execute(pool).await.expect("release").last_insert_rowid();
        sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)")
            .bind(movie_id)
            .bind(release_id)
            .execute(pool)
            .await
            .expect("association");
        let releases: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/movies/{movie_id}/releases"))
                .await
                .expect("releases")
                .json()
                .await
                .expect("release json");
        assert_eq!(releases[0]["verdict"], "eligible");

        let response = reqwest::Client::new()
            .post(format!(
                "{base}/api/movies/{movie_id}/releases/{release_id}/grab"
            ))
            .send()
            .await
            .expect("grab");
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        assert_eq!(
            state.next_movie_command().await.expect("command"),
            MovieCommand::Grab {
                movie_id,
                release_id
            }
        );
    }

    /// #114: a deck must score against the current policy, not whatever the
    /// score was when the release was last searched — tuning must reach what
    /// the user sees without a re-search.
    #[tokio::test]
    async fn the_deck_rescores_against_the_current_policy_not_the_stored_score() {
        let (_dir, state, base) = application().await;
        let movie = add_movie(&base, 693_134, 1).await;
        let movie_id = movie["id"].as_i64().expect("id");
        let pool = state.database().expect("database").pool();

        let name = "Dune Part Two 2024 1080p BluRay x264-GROUP";
        let parsed = arr_parse::parse(name);
        let size_bytes: u64 = 20 * (1 << 30);
        let seeders: u32 = 50;
        let size: i64 = i64::try_from(size_bytes).expect("size fits i64");

        let release_id = sqlx::query(
            "INSERT INTO releases (indexer_id, guid, name, size, seeders, download_url, parsed, score, verdict)
             VALUES (7, 'guid', ?, ?, ?, 'url', ?, -12345, 'eligible')",
        )
        .bind(name)
        .bind(size)
        .bind(i64::from(seeders))
        .bind(serde_json::to_string(&parsed).expect("parsed json"))
        .execute(pool)
        .await
        .expect("release")
        .last_insert_rowid();
        sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)")
            .bind(movie_id)
            .bind(release_id)
            .execute(pool)
            .await
            .expect("association");

        let policy = state
            .database()
            .expect("database")
            .movie_policy(movie_id)
            .await
            .expect("policy")
            .expect("movie has a policy")
            .policy;
        let expected = arr_core::score::score(
            &policy,
            arr_core::policy::Candidate::PreGrab(&parsed),
            size_bytes,
            seeders,
        )
        .total;
        #[allow(clippy::cast_precision_loss)]
        let expected = expected as f64;

        let releases: Vec<serde_json::Value> =
            reqwest::get(format!("{base}/api/movies/{movie_id}/releases"))
                .await
                .expect("releases")
                .json()
                .await
                .expect("release json");

        assert_ne!(releases[0]["score"].as_f64(), Some(-12345.0));
        assert_eq!(releases[0]["score"].as_f64(), Some(expected));
    }

    #[tokio::test]
    async fn a_full_action_queue_returns_service_unavailable() {
        let (_dir, state, base) = application().await;
        let movie = add_movie(&base, 693_134, 2).await;
        let movie_id = movie["id"].as_i64().expect("id");
        for _ in 0..64 {
            state
                .send_movie_command(MovieCommand::Search { movie_id })
                .expect("queue has capacity");
        }

        let response = reqwest::Client::new()
            .post(format!("{base}/api/movies/{movie_id}/search"))
            .send()
            .await
            .expect("search");
        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
    }

    #[tokio::test]
    async fn attention_queues_follow_search_and_failure_state() {
        let (_dir, state, base) = application().await;
        let pool = state.database().expect("database").pool();
        let root_id: i64 = sqlx::query_scalar("SELECT id FROM roots WHERE audience = 'kids'")
            .fetch_one(pool)
            .await
            .expect("kids root");
        let movie = add_movie(&base, 82728, root_id).await;
        let movie_id = movie["id"].as_i64().expect("id");
        sqlx::query("UPDATE movies SET search_attempts = 1 WHERE id = ?")
            .bind(movie_id)
            .execute(pool)
            .await
            .expect("record search");

        let mut release_ids = Vec::new();
        for suffix in ["a", "b"] {
            let release_id = sqlx::query("INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict, rejected_rule) VALUES (1, ?, ?, 1, 'url', '{}', 'rejected', 'required_audio')")
                .bind(suffix)
                .bind(suffix)
                .execute(pool)
                .await
                .expect("release")
                .last_insert_rowid();
            release_ids.push(release_id);
            sqlx::query("INSERT INTO grabs (release_id, target_kind, target_id, infohash, state) VALUES (?, 'movie', ?, ?, 'failed')")
                .bind(release_id)
                .bind(movie_id)
                .bind(format!("hash-{suffix}"))
                .execute(pool)
                .await
                .expect("failed grab");
        }

        let queues: serde_json::Value = reqwest::get(format!("{base}/api/queues/attention"))
            .await
            .expect("queues")
            .json()
            .await
            .expect("queues json");
        assert_eq!(queues["no_pt_source"][0]["id"], movie_id);
        assert_eq!(queues["needs_decision"][0]["id"], movie_id);

        sqlx::query("UPDATE movies SET blocked = 1 WHERE id = ?")
            .bind(movie_id)
            .execute(pool)
            .await
            .expect("block movie");
        let queues: serde_json::Value = reqwest::get(format!("{base}/api/queues/attention"))
            .await
            .expect("blocked queues")
            .json()
            .await
            .expect("blocked queues json");
        assert_eq!(queues["no_pt_source"].as_array().map(Vec::len), Some(0));
        let search = reqwest::Client::new()
            .post(format!("{base}/api/movies/{movie_id}/search"))
            .send()
            .await
            .expect("blocked search");
        assert_eq!(search.status(), StatusCode::CONFLICT);

        sqlx::query("UPDATE movies SET blocked = 0 WHERE id = ?")
            .bind(movie_id)
            .execute(pool)
            .await
            .expect("unblock movie");

        sqlx::query("UPDATE releases SET verdict = 'eligible', rejected_rule = NULL WHERE id = ?")
            .bind(release_ids[0])
            .execute(pool)
            .await
            .expect("eligible release");
        sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)")
            .bind(movie_id)
            .bind(release_ids[0])
            .execute(pool)
            .await
            .expect("associate release");

        let queues: serde_json::Value = reqwest::get(format!("{base}/api/queues/attention"))
            .await
            .expect("queues")
            .json()
            .await
            .expect("queues json");
        assert_eq!(queues["no_pt_source"].as_array().map(Vec::len), Some(0));
        assert_eq!(queues["needs_decision"][0]["id"], movie_id);
    }
}