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

//! Policy CRUD (`DESIGN.md` §5.1, §10). Policies are DB rows tuned by hand;
//! this surface replaces editing SQL inside the container. Payloads mirror
//! the JSON columns the migrations store, validated against the shapes
//! `arr-core` evaluates — an unknown resolution or a malformed field is a
//! 422 naming the field, never a 500.

use std::collections::BTreeMap;

use axum::extract::rejection::JsonRejection;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::Json;
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;

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

/// Which audio track a release must carry (`DESIGN.md` §5.2), expressed
/// against the title's original language rather than a fixed list.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(tag = "require", rename_all = "snake_case")]
pub enum RequiredAudioSpec {
    OriginalLanguage,
    AnyOf { langs: Vec<String> },
}

/// Dolby Vision profiles rejected post-probe (`DESIGN.md` §5.3). Profile
/// numbers stay strings to match the stored column.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
pub struct HdrRulesSpec {
    #[serde(default)]
    pub dv_profile_reject: Vec<String>,
}

/// One resolution's size band (`DESIGN.md` §5.5), sizes in gibibytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
pub struct SizeBandSpec {
    pub floor_gib: u64,
    pub target_gib: u64,
    pub penalty_points_per_gib_over: i32,
}

/// The scoring weights (`DESIGN.md` §5.5). `resolution_step` defaults to the
/// engine's own value when absent, matching how `arr-db` reads older rows.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
pub struct ScoreWeightsSpec {
    pub size_at_target: i32,
    pub source_tier: i32,
    pub seeder_doubling: i32,
    #[serde(default = "default_resolution_step")]
    pub resolution_step: i32,
}

fn default_resolution_step() -> i32 {
    arr_core::ScoreWeights::default().resolution_step
}

/// A full policy document — one row of the `policies` table.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)]
pub struct Policy {
    pub id: i64,
    pub name: String,
    pub required_audio: RequiredAudioSpec,
    pub dub_blacklist: Vec<String>,
    pub hdr_rules: HdrRulesSpec,
    pub size_bands: BTreeMap<String, SizeBandSpec>,
    pub resolution_pref: Vec<String>,
    pub source_weights: BTreeMap<String, i32>,
    pub score_weights: ScoreWeightsSpec,
}

/// The payload for creating or replacing a policy. Same shape as [`Policy`]
/// minus the id.
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct PolicyInput {
    pub name: String,
    pub required_audio: RequiredAudioSpec,
    #[serde(default)]
    pub dub_blacklist: Vec<String>,
    #[serde(default)]
    pub hdr_rules: HdrRulesSpec,
    #[serde(default)]
    pub size_bands: BTreeMap<String, SizeBandSpec>,
    #[serde(default)]
    pub resolution_pref: Vec<String>,
    #[serde(default)]
    pub source_weights: BTreeMap<String, i32>,
    pub score_weights: ScoreWeightsSpec,
}

impl PolicyInput {
    /// Validate against the vocabulary the policy engine knows. Every
    /// failure names its field so a rejected edit is fixable without
    /// reading the schema.
    fn validate(&self) -> Result<(), String> {
        if self.name.trim().is_empty() {
            return Err("name: must not be empty".into());
        }
        if let RequiredAudioSpec::AnyOf { langs } = &self.required_audio {
            if langs.is_empty() {
                return Err("required_audio.langs: any_of needs at least one language".into());
            }
            if langs.iter().any(String::is_empty) {
                return Err("required_audio.langs: language tags must not be empty".into());
            }
        }
        for lang in &self.dub_blacklist {
            if lang.is_empty() {
                return Err("dub_blacklist: language tags must not be empty".into());
            }
        }
        for profile in &self.hdr_rules.dv_profile_reject {
            if profile.parse::<u8>().is_err() {
                return Err(format!(
                    "hdr_rules.dv_profile_reject: '{profile}' is not a profile number"
                ));
            }
        }
        for resolution in self.size_bands.keys() {
            if arr_db::policy::resolution_value(resolution).is_none() {
                return Err(format!("size_bands: unknown resolution '{resolution}'"));
            }
        }
        let mut seen = std::collections::BTreeSet::new();
        for resolution in &self.resolution_pref {
            if arr_db::policy::resolution_value(resolution).is_none() {
                return Err(format!(
                    "resolution_pref: unknown resolution '{resolution}'"
                ));
            }
            if !seen.insert(resolution.as_str()) {
                return Err(format!("resolution_pref: '{resolution}' appears twice"));
            }
        }
        for source in self.source_weights.keys() {
            if arr_db::policy::source_value(source).is_none() {
                return Err(format!("source_weights: unknown source tier '{source}'"));
            }
        }
        Ok(())
    }

    fn into_columns(self, name: String) -> Result<PolicyColumns, ApiError> {
        fn json(value: impl serde::Serialize) -> Result<String, ApiError> {
            serde_json::to_string(&value).map_err(|error| {
                tracing::error!(%error, "policy serialisation failed");
                ApiError::Database("serialisation failed".into())
            })
        }
        Ok(PolicyColumns {
            name,
            required_audio: json(&self.required_audio)?,
            dub_blacklist: json(self.dub_blacklist)?,
            hdr_rules: json(self.hdr_rules)?,
            size_bands: json(self.size_bands)?,
            resolution_pref: json(self.resolution_pref)?,
            source_weights: json(self.source_weights)?,
            score_weights: json(self.score_weights)?,
        })
    }
}

/// A policy row as the table stores it, before JSON parsing.
struct PolicyColumns {
    name: String,
    required_audio: String,
    dub_blacklist: String,
    hdr_rules: String,
    size_bands: String,
    resolution_pref: String,
    source_weights: String,
    score_weights: String,
}

fn column<T: serde::de::DeserializeOwned>(
    column: &'static str,
    value: &str,
) -> Result<T, ApiError> {
    serde_json::from_str(value).map_err(|error| {
        tracing::error!(column, %error, "policy column holds unexpected JSON");
        ApiError::Database(format!("policy column {column} is not valid"))
    })
}

impl PolicyColumns {
    fn into_policy(self, id: i64) -> Result<Policy, ApiError> {
        Ok(Policy {
            id,
            name: self.name,
            required_audio: column("required_audio", &self.required_audio)?,
            dub_blacklist: column("dub_blacklist", &self.dub_blacklist)?,
            hdr_rules: column("hdr_rules", &self.hdr_rules)?,
            size_bands: column("size_bands", &self.size_bands)?,
            resolution_pref: column("resolution_pref", &self.resolution_pref)?,
            source_weights: column("source_weights", &self.source_weights)?,
            score_weights: column("score_weights", &self.score_weights)?,
        })
    }
}

/// The extracted body, whether or not it parsed. Both malformed JSON and a
/// shape mismatch answer 422 with the parser's own field-naming message.
pub(crate) fn parsed<T>(payload: Result<Json<T>, JsonRejection>) -> Result<T, ApiError> {
    payload
        .map(|Json(value)| value)
        .map_err(|rejection| ApiError::Invalid(rejection.body_text()))
}

fn is_unique_violation(error: &sqlx::Error) -> bool {
    error
        .as_database_error()
        .is_some_and(sqlx::error::DatabaseError::is_unique_violation)
}

async fn load_policy(state: &AppState, id: i64) -> Result<Policy, ApiError> {
    let row = sqlx::query_as!(
        PolicyColumns,
        r#"SELECT name AS "name!: String",
                  required_audio  AS "required_audio!: String",
                  dub_blacklist   AS "dub_blacklist!: String",
                  hdr_rules       AS "hdr_rules!: String",
                  size_bands      AS "size_bands!: String",
                  resolution_pref AS "resolution_pref!: String",
                  source_weights  AS "source_weights!: String",
                  score_weights   AS "score_weights!: String"
           FROM policies WHERE id = ?"#,
        id
    )
    .fetch_optional(pool(state)?)
    .await?
    .ok_or(ApiError::PolicyNotFound)?;
    row.into_policy(id)
}

#[utoipa::path(
    get, path = "/api/policies", tag = "policies",
    responses(
        (status = 200, body = [Policy]),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn list(State(state): State<AppState>) -> Result<Json<Vec<Policy>>, ApiError> {
    let rows = sqlx::query_as!(
        PolicyRow,
        r#"SELECT id AS "id!: i64",
                  name            AS "name!: String",
                  required_audio  AS "required_audio!: String",
                  dub_blacklist   AS "dub_blacklist!: String",
                  hdr_rules       AS "hdr_rules!: String",
                  size_bands      AS "size_bands!: String",
                  resolution_pref AS "resolution_pref!: String",
                  source_weights  AS "source_weights!: String",
                  score_weights   AS "score_weights!: String"
           FROM policies ORDER BY id"#
    )
    .fetch_all(pool(&state)?)
    .await?;
    let mut policies = Vec::with_capacity(rows.len());
    for row in rows {
        policies.push(row.into_policy()?);
    }
    Ok(Json(policies))
}

#[derive(Debug, sqlx::FromRow)]
struct PolicyRow {
    id: i64,
    name: String,
    required_audio: String,
    dub_blacklist: String,
    hdr_rules: String,
    size_bands: String,
    resolution_pref: String,
    source_weights: String,
    score_weights: String,
}

impl PolicyRow {
    fn into_policy(self) -> Result<Policy, ApiError> {
        PolicyColumns {
            name: self.name,
            required_audio: self.required_audio,
            dub_blacklist: self.dub_blacklist,
            hdr_rules: self.hdr_rules,
            size_bands: self.size_bands,
            resolution_pref: self.resolution_pref,
            source_weights: self.source_weights,
            score_weights: self.score_weights,
        }
        .into_policy(self.id)
    }
}

#[utoipa::path(
    get, path = "/api/policies/{policy_id}", tag = "policies",
    params(("policy_id" = i64, Path, description = "Policy row id")),
    responses(
        (status = 200, body = Policy),
        (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<Policy>, ApiError> {
    Ok(Json(load_policy(&state, id).await?))
}

#[utoipa::path(
    post, path = "/api/policies", tag = "policies", request_body = PolicyInput,
    responses(
        (status = 201, body = Policy),
        (status = 409, body = ErrorBody),
        (status = 422, body = ErrorBody),
        (status = 500, body = ErrorBody),
        (status = 503, body = ErrorBody)
    )
)]
pub async fn create(
    State(state): State<AppState>,
    body: Result<Json<PolicyInput>, JsonRejection>,
) -> Result<(StatusCode, Json<Policy>), ApiError> {
    let input = parsed(body)?;
    input.validate().map_err(ApiError::Invalid)?;
    let name = input.name.trim().to_owned();
    let columns = input.into_columns(name)?;
    let result = sqlx::query!(
        r#"INSERT INTO policies (
               name, required_audio, dub_blacklist, hdr_rules,
               size_bands, resolution_pref, source_weights, score_weights
           ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"#,
        columns.name,
        columns.required_audio,
        columns.dub_blacklist,
        columns.hdr_rules,
        columns.size_bands,
        columns.resolution_pref,
        columns.source_weights,
        columns.score_weights,
    )
    .execute(pool(&state)?)
    .await
    .map_err(|error| {
        if is_unique_violation(&error) {
            ApiError::Conflict("a policy with this name already exists".into())
        } else {
            ApiError::from(error)
        }
    })?;
    let policy = load_policy(&state, result.last_insert_rowid()).await?;
    Ok((StatusCode::CREATED, Json(policy)))
}

#[utoipa::path(
    put, path = "/api/policies/{policy_id}", tag = "policies", request_body = PolicyInput,
    params(("policy_id" = i64, Path, description = "Policy row id")),
    responses(
        (status = 200, body = Policy),
        (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>,
    body: Result<Json<PolicyInput>, JsonRejection>,
) -> Result<Json<Policy>, ApiError> {
    let input = parsed(body)?;
    input.validate().map_err(ApiError::Invalid)?;
    let name = input.name.trim().to_owned();
    let columns = input.into_columns(name)?;
    let result = sqlx::query!(
        r#"UPDATE policies SET
               name = ?, required_audio = ?, dub_blacklist = ?, hdr_rules = ?,
               size_bands = ?, resolution_pref = ?, source_weights = ?,
               score_weights = ?,
               updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
           WHERE id = ?"#,
        columns.name,
        columns.required_audio,
        columns.dub_blacklist,
        columns.hdr_rules,
        columns.size_bands,
        columns.resolution_pref,
        columns.source_weights,
        columns.score_weights,
        id,
    )
    .execute(pool(&state)?)
    .await
    .map_err(|error| {
        if is_unique_violation(&error) {
            ApiError::Conflict("a policy with this name already exists".into())
        } else {
            ApiError::from(error)
        }
    })?;
    if result.rows_affected() == 0 {
        return Err(ApiError::PolicyNotFound);
    }
    Ok(Json(load_policy(&state, id).await?))
}

#[utoipa::path(
    delete, path = "/api/policies/{policy_id}", tag = "policies",
    params(("policy_id" = i64, Path, description = "Policy 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> {
    let references: i64 = sqlx::query_scalar!(
        r#"SELECT count(*) AS "count!: i64" FROM roots WHERE policy_id = ?"#,
        id
    )
    .fetch_one(pool(&state)?)
    .await?;
    if references > 0 {
        return Err(ApiError::Conflict(format!(
            "this policy is attached to {references} root{} — detach it first",
            if references == 1 { "" } else { "s" }
        )));
    }
    let result = sqlx::query!("DELETE FROM policies WHERE id = ?", id)
        .execute(pool(&state)?)
        .await?;
    if result.rows_affected() == 0 {
        return Err(ApiError::PolicyNotFound);
    }
    Ok(StatusCode::NO_CONTENT)
}

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

    async fn application() -> (tempfile::TempDir, 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);
        tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
        (dir, format!("http://{address}"))
    }

    fn valid_input(name: &str) -> serde_json::Value {
        serde_json::json!({
            "name": name,
            "required_audio": { "require": "any_of", "langs": ["pt-PT"] },
            "dub_blacklist": ["pt-BR"],
            "hdr_rules": { "dv_profile_reject": ["5", "7"] },
            "size_bands": {
                "2160p": { "floor_gib": 8, "target_gib": 22, "penalty_points_per_gib_over": 60 },
                "1080p": { "floor_gib": 3, "target_gib": 8, "penalty_points_per_gib_over": 60 }
            },
            "resolution_pref": ["2160p", "1080p"],
            "source_weights": { "WEB-DL": 4, "Remux": 1 },
            "score_weights": {
                "size_at_target": 1000, "source_tier": 25,
                "seeder_doubling": 8, "resolution_step": 300
            }
        })
    }

    async fn create(base: &str, payload: serde_json::Value) -> reqwest::Response {
        reqwest::Client::new()
            .post(format!("{base}/api/policies"))
            .json(&payload)
            .send()
            .await
            .expect("create policy")
    }

    #[tokio::test]
    async fn crud_round_trips_a_policy() {
        let (_dir, base) = application().await;

        let created: serde_json::Value = create(&base, valid_input("test policy"))
            .await
            .json()
            .await
            .expect("created json");
        assert_eq!(created["name"], "test policy");
        assert_eq!(created["required_audio"]["require"], "any_of");
        assert_eq!(created["score_weights"]["resolution_step"], 300);
        let id = created["id"].as_i64().expect("id");

        let listed: Vec<serde_json::Value> = reqwest::get(format!("{base}/api/policies"))
            .await
            .expect("list")
            .json()
            .await
            .expect("list json");
        assert_eq!(listed.len(), 5, "four seeded policies plus the new one");

        let fetched: serde_json::Value = reqwest::get(format!("{base}/api/policies/{id}"))
            .await
            .expect("get")
            .json()
            .await
            .expect("get json");
        assert_eq!(fetched, created);

        let mut replacement = valid_input("renamed");
        replacement["resolution_pref"] = serde_json::json!(["1080p"]);
        let updated: serde_json::Value = reqwest::Client::new()
            .put(format!("{base}/api/policies/{id}"))
            .json(&replacement)
            .send()
            .await
            .expect("update")
            .json()
            .await
            .expect("updated json");
        assert_eq!(updated["name"], "renamed");
        assert_eq!(updated["resolution_pref"], serde_json::json!(["1080p"]));

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

    #[tokio::test]
    async fn a_referenced_policy_refuses_to_die() {
        let (_dir, base) = application().await;
        let roots: Vec<serde_json::Value> = reqwest::get(format!("{base}/api/roots"))
            .await
            .expect("roots")
            .json()
            .await
            .expect("roots json");
        let policy_id = roots[0]["policy_id"].as_i64().expect("policy id");

        let response = reqwest::Client::new()
            .delete(format!("{base}/api/policies/{policy_id}"))
            .send()
            .await
            .expect("delete referenced policy");

        assert_eq!(response.status(), StatusCode::CONFLICT);
        let body: serde_json::Value = response.json().await.expect("error body");
        assert!(
            body["error"].as_str().expect("error text").contains("root"),
            "the error names the problem: {body}"
        );
    }

    #[tokio::test]
    async fn an_unknown_resolution_is_a_422_naming_the_field() {
        let (_dir, base) = application().await;
        let mut payload = valid_input("bad bands");
        payload["size_bands"]["1440p"] = serde_json::json!({ "floor_gib": 2, "target_gib": 6, "penalty_points_per_gib_over": 60 });

        let response = create(&base, payload).await;
        assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
        let body: serde_json::Value = response.json().await.expect("error body");
        let error = body["error"].as_str().expect("error text");
        assert!(error.contains("size_bands"), "{error}");
        assert!(error.contains("1440p"), "{error}");
    }

    #[tokio::test]
    async fn every_field_validates_by_name() {
        let (_dir, base) = application().await;
        let with = |patch: &dyn Fn(&mut serde_json::Value)| {
            let mut payload = valid_input("validation probe");
            patch(&mut payload);
            payload
        };
        let cases: Vec<(serde_json::Value, &str)> = vec![
            (
                with(&|payload| payload["name"] = serde_json::json!("")),
                "name",
            ),
            (
                with(&|payload| {
                    payload["required_audio"] =
                        serde_json::json!({ "require": "any_of", "langs": [] });
                }),
                "required_audio",
            ),
            (
                with(&|payload| {
                    payload["hdr_rules"] = serde_json::json!({ "dv_profile_reject": ["nine"] });
                }),
                "dv_profile_reject",
            ),
            (
                with(&|payload| {
                    payload["resolution_pref"] = serde_json::json!(["2160p", "2160p"]);
                }),
                "resolution_pref",
            ),
            (
                with(&|payload| {
                    payload["source_weights"] = serde_json::json!({ "LaserDisc": 3 });
                }),
                "source_weights",
            ),
        ];

        for (payload, field) in cases {
            let response = create(&base, payload).await;
            assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
            let body: serde_json::Value = response.json().await.expect("error body");
            let error = body["error"].as_str().expect("error text");
            assert!(error.contains(field), "{field}: {error}");
        }
    }

    #[tokio::test]
    async fn malformed_json_is_422_not_400_or_500() {
        let (_dir, base) = application().await;
        let client = reqwest::Client::new();
        let response = client
            .post(format!("{base}/api/policies"))
            .header("content-type", "application/json")
            .body("{not json")
            .send()
            .await
            .expect("malformed create");
        assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);

        let response = client
            .post(format!("{base}/api/policies"))
            .json(&serde_json::json!({ "name": 42 }))
            .send()
            .await
            .expect("wrong-shape create");
        assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
    }

    #[tokio::test]
    async fn a_duplicate_name_conflicts() {
        let (_dir, base) = application().await;
        let response = create(&base, valid_input("Movies — main")).await;
        assert_eq!(response.status(), StatusCode::CONFLICT);
    }
}