arr/crates/arr-api/src/reclassify.rs

//! Stored verdicts, re-derived when a title's overrides change (§9.3).
//!
//! A release's verdict is stamped once, by the search that found it. Both the
//! deck and the daemon's manual-grab gate read that stored column, so an
//! override written from the deck's one-click waive would change nothing
//! until the next sweep — the row the operator just acted on would keep
//! reading `rejected` and the grab would be refused.
//!
//! So the rules run again here, over the releases already attached to the
//! title. This is the same correction the daemon makes when a grab turns out
//! to be blacklisted: fix the row now rather than leave §9.3 offering, or
//! refusing, something that is no longer true.
//!
//! The blacklist (§6.3) is not a policy opinion, so a row rejected by it is
//! left exactly as it is.

use std::collections::BTreeMap;

use arr_core::policy::{evaluate, Candidate};
use arr_core::score::claimed_episode_count;
use arr_core::{Language, ParsedRelease, Policy, TitleOverrides};
use arr_db::blacklist;

use crate::movies::{pool, ApiError};
use crate::search::verdict_columns;
use crate::state::AppState;

/// One stored release, as much of it as re-evaluation needs.
struct Stored {
    id: i64,
    size: i64,
    parsed: serde_json::Value,
    verdict: Option<String>,
    rejected_rule: Option<String>,
}

/// Re-evaluate every release attached to one movie.
///
/// A movie with no stored original language is left alone, the same rule the
/// daemon's search applies: without it the required-audio rule has nothing to
/// compare, and guessing would move verdicts on worse evidence than the ones
/// already there.
pub(crate) async fn movie(state: &AppState, movie_id: i64) -> Result<(), ApiError> {
    let Some(language) = movie_language(state, movie_id).await? else {
        return Ok(());
    };
    let Some(loaded) = state
        .database()
        .ok_or(ApiError::Unavailable)?
        .movie_policy(movie_id)
        .await
        .map_err(|error| ApiError::Database(error.to_string()))?
    else {
        return Ok(());
    };
    let releases = sqlx::query_as!(
        Stored,
        r#"
        SELECT   r.id            AS "id!: i64",
                 r.size          AS "size!: i64",
                 r.parsed        AS "parsed!: serde_json::Value",
                 r.verdict,
                 r.rejected_rule
        FROM     releases r
        JOIN     movie_releases mr ON mr.release_id = r.id
        WHERE    mr.movie_id = ?
        "#,
        movie_id
    )
    .fetch_all(pool(state)?)
    .await?;

    apply(
        state,
        &releases,
        &loaded.policy,
        &loaded.overrides,
        &language,
        None,
        0,
    )
    .await
}

/// Re-evaluate every release attached to one series, episode decks and season
/// decks alike — overrides sit on the series (§5.1), so one change moves both.
pub(crate) async fn series(state: &AppState, series_id: i64) -> Result<(), ApiError> {
    let Some(language) = series_language(state, series_id).await? else {
        return Ok(());
    };
    let Some(loaded) = state
        .database()
        .ok_or(ApiError::Unavailable)?
        .series_policy(series_id)
        .await
        .map_err(|error| ApiError::Database(error.to_string()))?
    else {
        return Ok(());
    };
    let releases = sqlx::query_as!(
        Stored,
        r#"
        SELECT   r.id            AS "id!: i64",
                 r.size          AS "size!: i64",
                 r.parsed        AS "parsed!: serde_json::Value",
                 r.verdict,
                 r.rejected_rule
        FROM     releases r
        WHERE    r.id IN (
                     SELECT er.release_id
                     FROM   episode_releases er
                     JOIN   episodes e ON e.id = er.episode_id
                     JOIN   seasons se ON se.id = e.season_id
                     WHERE  se.series_id = ?
                     UNION
                     SELECT sr.release_id
                     FROM   season_releases sr
                     JOIN   seasons se ON se.id = sr.season_id
                     WHERE  se.series_id = ?
                 )
        "#,
        series_id,
        series_id
    )
    .fetch_all(pool(state)?)
    .await?;

    // A size band describes one episode (§5.5), so a pack's verdict needs the
    // same divisor the deck scores it with.
    let lengths = crate::series::season_lengths(state, series_id).await?;
    let runtime = crate::series::series_runtime(state, series_id).await?;
    apply(
        state,
        &releases,
        &loaded.policy,
        &loaded.overrides,
        &language,
        Some(&lengths),
        runtime,
    )
    .await
}

#[allow(clippy::too_many_arguments)]
async fn apply(
    state: &AppState,
    releases: &[Stored],
    policy: &Policy,
    overrides: &TitleOverrides,
    original_language: &Language,
    season_lengths: Option<&BTreeMap<u32, u32>>,
    runtime_minutes: u32,
) -> Result<(), ApiError> {
    for release in releases {
        if release.rejected_rule.as_deref() == Some(blacklist::RULE) {
            continue;
        }
        let parsed: ParsedRelease = serde_json::from_value(release.parsed.clone())
            .map_err(|error| ApiError::Database(error.to_string()))?;
        let episodes = season_lengths.map_or(1, |lengths| {
            claimed_episode_count(parsed.episode.as_ref(), lengths)
        });
        // A stored zero is a release the indexer gave no size for, and a
        // release with no size has nothing to say about its band — the same
        // reading the deck's rescore takes.
        let size = u64::try_from(release.size).ok().filter(|size| *size > 0);
        let evaluation = evaluate(
            policy,
            overrides,
            original_language,
            Candidate::PreGrab(&parsed),
            size,
            episodes,
            runtime_minutes,
        );
        // `releases` allows a rule name only on a rejected row
        // (`CHECK ((verdict = 'rejected') = (rejected_rule IS NOT NULL))`),
        // which is also how the daemon writes a waiver.
        let (verdict, rule) = verdict_columns(&evaluation.verdict);
        let rule = if verdict == "rejected" { rule } else { None };
        if release.verdict.as_deref() == Some(verdict) && release.rejected_rule == rule {
            continue;
        }
        sqlx::query!(
            "UPDATE releases SET verdict = ?, rejected_rule = ? WHERE id = ?",
            verdict,
            rule,
            release.id
        )
        .execute(pool(state)?)
        .await?;
    }
    Ok(())
}

async fn movie_language(state: &AppState, movie_id: i64) -> Result<Option<Language>, ApiError> {
    let stored = sqlx::query_scalar!(
        "SELECT original_language FROM movies WHERE id = ?",
        movie_id
    )
    .fetch_optional(pool(state)?)
    .await?
    .flatten();
    Ok(stored.map(|value| arr_db::policy::language(&value)))
}

async fn series_language(state: &AppState, series_id: i64) -> Result<Option<Language>, ApiError> {
    let stored = sqlx::query_scalar!(
        "SELECT original_language FROM series WHERE id = ?",
        series_id
    )
    .fetch_optional(pool(state)?)
    .await?
    .flatten();
    Ok(stored.map(|value| arr_db::policy::language(&value)))
}