JRay-public-server/src/api/fetch.rs

//! Manifest fetch endpoints (§4).
//!
//! §7: the submitted JSON was parsed, validated, resolved to TMDB person ids,
//! written as rows and discarded. Everything served here is **reconstructed**
//! from those rows, never echoed — which is what makes §5a's Threat 1 defence
//! structural rather than a promise.

use axum::extract::{Path, Query, State};
use axum::http::HeaderMap;
use axum::response::{IntoResponse, Response};
use axum::Json;
use serde::Serialize;

use super::LookupParams;
use crate::db::repo::{self, ManifestRow};
use crate::error::{ApiError, ApiResult};
use crate::matching::{self, StoredCut};
use crate::model::{
    Actor, Coverage, Cut, Extraction, GalleryScope, Identity, IdentityType, Jmanifest, MatchTier,
    SeriesBundle, SeriesRef, JMANIFEST_VERSION,
};
use crate::ratelimit::Surface;
use crate::state::{with_quota_headers, AppState};

#[derive(Debug, Serialize)]
pub struct FetchResponse {
    pub r#match: &'static str,
    /// Scene offset the client must add (§3). Zero for the tiers currently
    /// served; present unconditionally so the plugin contract does not change
    /// when `audio` is enabled.
    pub offset_sec: f64,
    pub manifest: Jmanifest,
}

#[derive(Debug, Serialize)]
pub struct StatusResponse {
    pub status: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

pub async fn get_movie(
    State(state): State<AppState>,
    peer: crate::state::PeerIp,
    headers: HeaderMap,
    Query(params): Query<LookupParams>,
) -> ApiResult<Response> {
    let ip = state.client_ip(&headers, peer.0);
    let quota = state.check_limit(&ip, Surface::ManifestFetch)?;

    if params.tmdb_id.is_none() && params.imdb_id.is_none() {
        return Err(ApiError::BadRequest("requires tmdb_id or imdb_id".into()));
    }
    let body = fetch_best(&state, IdentityType::Movie, &params, None, None).await?;
    Ok(with_quota_headers(Json(body).into_response(), quota))
}

pub async fn get_episode(
    State(state): State<AppState>,
    peer: crate::state::PeerIp,
    headers: HeaderMap,
    Query(params): Query<LookupParams>,
) -> ApiResult<Response> {
    let ip = state.client_ip(&headers, peer.0);
    let quota = state.check_limit(&ip, Surface::ManifestFetch)?;

    if params.series_tmdb_id.is_none() && params.series_imdb_id.is_none() {
        return Err(ApiError::BadRequest("requires series_tmdb_id or series_imdb_id".into()));
    }
    let (Some(season), Some(episode)) = (params.season, params.episode) else {
        return Err(ApiError::BadRequest("requires season and episode".into()));
    };
    let body =
        fetch_best(&state, IdentityType::Episode, &params, Some(season), Some(episode)).await?;
    Ok(with_quota_headers(Json(body).into_response(), quota))
}

async fn fetch_best(
    state: &AppState,
    kind: IdentityType,
    params: &LookupParams,
    season: Option<i64>,
    episode: Option<i64>,
) -> ApiResult<FetchResponse> {
    let (tmdb_id, imdb_id) = match kind {
        IdentityType::Movie => (params.tmdb_id.clone(), params.imdb_id.clone()),
        IdentityType::Episode => (params.series_tmdb_id.clone(), params.series_imdb_id.clone()),
    };
    let client_cut = params.client_cut();

    let found = state
        .db
        .read(move |conn| {
            let Some(title) = repo::find_title(conn, kind, tmdb_id.as_deref(), imdb_id.as_deref())?
            else {
                return Ok(None);
            };
            let candidates = repo::candidates_for_title(conn, &title.id, season, episode)?;
            let cuts: Vec<(ManifestRow, StoredCut)> = candidates
                .into_iter()
                .map(|m| {
                    let cut =
                        StoredCut { runtime_sec: m.runtime_sec, video_hash: m.video_hash.clone() };
                    (m, cut)
                })
                .collect();

            let Some((row, m)) = matching::best_match(&client_cut, &cuts) else {
                return Ok(None);
            };
            let manifest = reconstruct(conn, &row, &title, kind)?;
            Ok(Some((m.tier, m.offset_sec, manifest)))
        })
        .await
        .map_err(ApiError::Internal)?;

    // §4: `404` if none clears `loose`.
    let (tier, offset_sec, manifest) = found.ok_or(ApiError::NotFound)?;
    Ok(FetchResponse { r#match: tier.as_str(), offset_sec, manifest })
}

/// `GET /manifests/series/{series_tmdb_id}?season=` (§4).
///
/// Returns whatever episodes the server holds. **Partial bundles are normal** — a
/// bundle with 9 of 13 episodes is a valid, useful response, not an error (§2).
/// Episode-level cut matching is done client-side against the returned bundle,
/// since a client pulling a whole series already knows its own runtimes.
/// TRACES: UR-006 | PR-006
pub async fn get_series(
    State(state): State<AppState>,
    peer: crate::state::PeerIp,
    headers: HeaderMap,
    Path(series_tmdb_id): Path<String>,
    Query(params): Query<LookupParams>,
) -> ApiResult<Response> {
    let ip = state.client_ip(&headers, peer.0);
    let quota = state.check_limit(&ip, Surface::SeriesFetch)?;

    let season = params.season;
    let bundle = state
        .db
        .read(move |conn| {
            let Some(title) =
                repo::find_title(conn, IdentityType::Episode, Some(&series_tmdb_id), None)?
            else {
                return Ok(None);
            };
            let rows = repo::episodes_for_series(conn, &title.id, season)?;

            // Multiple contributors may hold the same episode; `episodes_for_series`
            // orders by rank, so keep the first per (season, episode).
            let mut episodes: Vec<Jmanifest> = Vec::new();
            let mut seen: Vec<(i64, i64)> = Vec::new();
            let mut seasons: Vec<i64> = Vec::new();

            for row in rows {
                let key = (row.season.unwrap_or(-1), row.episode.unwrap_or(-1));
                if seen.contains(&key) {
                    continue;
                }
                seen.push(key);
                if !seasons.contains(&key.0) {
                    seasons.push(key.0);
                }
                episodes.push(reconstruct(conn, &row, &title, IdentityType::Episode)?);
            }
            seasons.sort_unstable();

            Ok(Some(SeriesBundle {
                jmanifest_version: JMANIFEST_VERSION,
                series: SeriesRef {
                    series_tmdb_id: title.tmdb_id.clone(),
                    series_imdb_id: title.imdb_id.clone(),
                    title: title.name.clone(),
                },
                coverage: Some(Coverage { episodes_available: episodes.len(), seasons }),
                episodes,
            }))
        })
        .await
        .map_err(ApiError::Internal)?;

    let bundle = bundle.filter(|b| !b.episodes.is_empty()).ok_or(ApiError::NotFound)?;
    Ok(with_quota_headers(Json(bundle).into_response(), quota))
}

/// `GET /manifests/{id}` — fetch a specific manifest by its server-assigned id,
/// for debugging and for the "report this manifest" flow (§4).
pub async fn get_by_id(
    State(state): State<AppState>,
    peer: crate::state::PeerIp,
    headers: HeaderMap,
    Path(id): Path<String>,
) -> ApiResult<Response> {
    let ip = state.client_ip(&headers, peer.0);
    let quota = state.check_limit(&ip, Surface::ManifestFetch)?;

    let manifest = state
        .db
        .read(move |conn| {
            let Some(row) = repo::manifest_by_id(conn, &id)? else { return Ok(None) };
            // Unlisted manifests are not served to anyone (§6 stage 3).
            if row.status != "listed" && row.status != "flagged" {
                return Ok(None);
            }
            let title = title_of(conn, &row.title_id)?;
            let kind =
                if title.kind == "movie" { IdentityType::Movie } else { IdentityType::Episode };
            Ok(Some(reconstruct(conn, &row, &title, kind)?))
        })
        .await
        .map_err(ApiError::Internal)?;

    let manifest = manifest.ok_or(ApiError::NotFound)?;
    Ok(with_quota_headers(Json(manifest).into_response(), quota))
}

/// `GET /manifests/{id}/status` — poll the outcome of the asynchronous cast
/// check (§4).
pub async fn get_status(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> ApiResult<Json<StatusResponse>> {
    let found = state
        .db
        .read(move |conn| repo::manifest_status(conn, &id))
        .await
        .map_err(ApiError::Internal)?;

    match found {
        Some((status, reason)) => Ok(Json(StatusResponse { status, reason })),
        // §6 deletes rejected manifests, so a vanished id is reported as
        // rejected rather than as a bad route.
        None => Ok(Json(StatusResponse {
            status: "rejected".into(),
            reason: Some("not_found_or_rejected".into()),
        })),
    }
}

fn title_of(conn: &rusqlite::Connection, title_id: &str) -> anyhow::Result<repo::TitleRow> {
    let row = conn.query_row(
        "SELECT id, kind, tmdb_id, imdb_id, name, year, adult, certification
           FROM titles WHERE id = ?1",
        rusqlite::params![title_id],
        |r| {
            Ok(repo::TitleRow {
                id: r.get(0)?,
                kind: r.get(1)?,
                tmdb_id: r.get(2)?,
                imdb_id: r.get(3)?,
                name: r.get(4)?,
                year: r.get(5)?,
                adult: r.get::<_, i64>(6)? != 0,
                certification: r.get(7)?,
            })
        },
    )?;
    Ok(row)
}

/// Rebuilds a Jmanifest from stored rows.
///
/// Names come from `people` — populated from TMDB by the server — so `name` is
/// server-authoritative on download and a name a contributor invented does not
/// round-trip (§2, §5a).
/// TRACES: UR-010, UR-013 | DR-002 | SR-001, SR-002
pub fn reconstruct(
    conn: &rusqlite::Connection,
    row: &ManifestRow,
    title: &repo::TitleRow,
    kind: IdentityType,
) -> anyhow::Result<Jmanifest> {
    let stored = repo::actors_for_manifest(conn, &row.id)?;

    let actors = stored
        .into_iter()
        .map(|a| Actor {
            name: a.name,
            imdb_id: None,
            tmdb_id: Some(a.tmdb_person_id.to_string()),
            scenes: a
                .scenes_cs
                .into_iter()
                .map(|(s, e)| [s as f64 / 100.0, e as f64 / 100.0])
                .collect(),
        })
        .collect();

    let identity = match kind {
        IdentityType::Movie => Identity {
            kind,
            tmdb_id: title.tmdb_id.clone(),
            imdb_id: title.imdb_id.clone(),
            series_tmdb_id: None,
            series_imdb_id: None,
            season: None,
            episode: None,
            title: title.name.clone(),
            year: title.year,
        },
        IdentityType::Episode => Identity {
            kind,
            tmdb_id: None,
            imdb_id: None,
            series_tmdb_id: title.tmdb_id.clone(),
            series_imdb_id: title.imdb_id.clone(),
            season: row.season,
            episode: row.episode,
            title: title.name.clone(),
            year: title.year,
        },
    };

    // An unrecognised stored scope is served as absent rather than guessed at:
    // the column is written from a closed enum, so anything else means the row
    // predates a schema change and its meaning is unknown (UR-014's spirit).
    let gallery_scope = match row.gallery_scope.as_deref() {
        Some("global") => Some(GalleryScope::Global),
        Some("limited") => Some(GalleryScope::Limited),
        _ => None,
    };

    let extraction = Extraction {
        sample_fps: row.sample_fps,
        extinction_sec: row.extinction_sec,
        pipeline_version: row.pipeline_version.clone(),
        gallery_size: None,
        gallery_scope,
    };
    let has_extraction = extraction.sample_fps.is_some()
        || extraction.extinction_sec.is_some()
        || extraction.pipeline_version.is_some()
        || extraction.gallery_scope.is_some();

    Ok(Jmanifest {
        jmanifest_version: JMANIFEST_VERSION,
        identity,
        cut: Cut {
            runtime_sec: row.runtime_sec,
            container_duration_sec: None,
            video_hash: row.video_hash.clone(),
            audio_signature: None,
        },
        extraction: has_extraction.then_some(extraction),
        actors,
    })
}

/// Exposed so tests can assert the served tier strings.
pub fn tier_name(t: MatchTier) -> &'static str {
    t.as_str()
}