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

//! Contribution endpoints (§4) — UR-2 and UR-6.
//!
//! Both require a token (§5). Both return `202`: the upload has passed size and
//! schema validation and is held unlisted pending the asynchronous TMDB cast
//! check (§6 stage 3).

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

use crate::db::repo;
use crate::error::{ApiError, ApiResult};
use crate::ingest::{self, IngestOutcome};
use crate::model::{IdentityType, Jmanifest, SeriesBundle};
use crate::ratelimit::Surface;
use crate::state::{with_quota_headers, AppState};
use crate::validate::{self, limits};
use crate::worker::now_iso;

#[derive(Debug, Serialize)]
pub struct UploadAccepted {
    pub manifest_id: String,
    pub status: &'static str,
}

/// `POST /manifests` — UR-2.
/// TRACES: UR-002, UR-003 | SR-004 | PR-006
pub async fn post_manifest(
    State(state): State<AppState>,
    headers: HeaderMap,
    super::json::Json(manifest): super::json::Json<Jmanifest>,
) -> ApiResult<Response> {
    let contributor = state.require_contributor(&headers).await?;
    // §5: limits are per token where one is present.
    let quota = state.check_limit(&contributor.id, Surface::ManifestUpload)?;

    // §6 stage 2. A rejection names the offending field, so a client that forgets
    // to strip `movie`/`jellyfin_id` gets a diagnosable `400`.
    let valid =
        validate::validate_manifest(manifest).map_err(|e| ApiError::BadRequest(e.to_string()))?;

    let origin = state.config.server_id.clone();
    let contributor_id = contributor.id.clone();
    let now = now_iso();

    let outcome = state
        .db
        .write(move |tx| ingest::persist(tx, &valid, Some(&contributor_id), &origin, None, &now))
        .await
        .map_err(ApiError::Internal)?;

    let resp = match outcome {
        IngestOutcome::Pending { manifest_id } => {
            (StatusCode::ACCEPTED, Json(UploadAccepted { manifest_id, status: "pending" }))
                .into_response()
        }
        // §4 `409` — an identical `(identity, cut)` manifest already exists from
        // this contributor.
        IngestOutcome::DuplicateFromContributor { manifest_id } => {
            return Err(ApiError::Conflict(format!(
                "an identical manifest already exists from this contributor: {manifest_id}"
            )))
        }
        // §9a: identical content already held, from any source. Not an error —
        // the contributor's work is simply already represented.
        IngestOutcome::DuplicateContent { manifest_id } => {
            (StatusCode::OK, Json(UploadAccepted { manifest_id, status: "already_present" }))
                .into_response()
        }
    };

    Ok(with_quota_headers(resp, quota))
}

#[derive(Debug, Serialize)]
pub struct BundleResult {
    pub season: Option<i64>,
    pub episode: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub manifest_id: Option<String>,
    pub status: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

#[derive(Debug, Serialize)]
pub struct BundleAccepted {
    pub results: Vec<BundleResult>,
}

/// `POST /manifests/bundle` — UR-6.
///
/// **Per-episode validation, not atomic**: valid episodes are accepted and
/// invalid ones rejected, with a per-episode result list. All-or-nothing would let
/// one bad episode discard an entire season's compute (§2).
///
/// **One rate-limit unit**, so contributing a season is not punished relative to
/// contributing a film (§2, §5).
/// TRACES: UR-006 | PR-006
pub async fn post_bundle(
    State(state): State<AppState>,
    headers: HeaderMap,
    super::json::Json(bundle): super::json::Json<SeriesBundle>,
) -> ApiResult<Response> {
    let contributor = state.require_contributor(&headers).await?;
    let quota = state.check_limit(&contributor.id, Surface::BundleUpload)?;

    // §4: `413` for exceeding the episode cap, distinct from a malformed envelope.
    if bundle.episodes.len() > limits::MAX_BUNDLE_EPISODES {
        return Err(ApiError::PayloadTooLarge(format!(
            "bundle carries {} episodes, limit is {}",
            bundle.episodes.len(),
            limits::MAX_BUNDLE_EPISODES
        )));
    }
    // §4: `400` only for the envelope itself; individual bad episodes are
    // reported in the results list, not as a whole-request error.
    validate::validate_bundle_envelope(&bundle).map_err(|e| ApiError::BadRequest(e.to_string()))?;

    let series_tmdb = bundle.series.series_tmdb_id.clone();
    let mut results = Vec::with_capacity(bundle.episodes.len());

    for episode in bundle.episodes {
        let coords = (episode.identity.season, episode.identity.episode);

        // An episode whose identity contradicts the envelope is rejected on its
        // own rather than being silently reattributed to the bundle's series.
        if episode.identity.kind != IdentityType::Episode {
            results.push(BundleResult {
                season: coords.0,
                episode: coords.1,
                manifest_id: None,
                status: "rejected",
                reason: Some("identity.type must be 'episode' within a bundle".into()),
            });
            continue;
        }
        if let (Some(envelope), Some(ep)) = (&series_tmdb, &episode.identity.series_tmdb_id) {
            if envelope != ep {
                results.push(BundleResult {
                    season: coords.0,
                    episode: coords.1,
                    manifest_id: None,
                    status: "rejected",
                    reason: Some("series_tmdb_id does not match the bundle envelope".into()),
                });
                continue;
            }
        }

        let valid = match validate::validate_manifest(episode) {
            Ok(v) => v,
            Err(e) => {
                results.push(BundleResult {
                    season: coords.0,
                    episode: coords.1,
                    manifest_id: None,
                    status: "rejected",
                    reason: Some(e.to_string()),
                });
                continue;
            }
        };

        let origin = state.config.server_id.clone();
        let contributor_id = contributor.id.clone();
        let now = now_iso();
        // One transaction per episode, so a bundle never holds the write lock for
        // the whole request (§8 chunked ingest reasoning).
        let outcome = state
            .db
            .write(move |tx| {
                ingest::persist(tx, &valid, Some(&contributor_id), &origin, None, &now)
            })
            .await;

        results.push(match outcome {
            Ok(IngestOutcome::Pending { manifest_id }) => BundleResult {
                season: coords.0,
                episode: coords.1,
                manifest_id: Some(manifest_id),
                status: "pending",
                reason: None,
            },
            Ok(IngestOutcome::DuplicateFromContributor { manifest_id })
            | Ok(IngestOutcome::DuplicateContent { manifest_id }) => BundleResult {
                season: coords.0,
                episode: coords.1,
                manifest_id: Some(manifest_id),
                status: "already_present",
                reason: None,
            },
            Err(e) => {
                tracing::error!(error = ?e, "bundle episode failed to persist");
                BundleResult {
                    season: coords.0,
                    episode: coords.1,
                    manifest_id: None,
                    status: "rejected",
                    reason: Some("internal error".into()),
                }
            }
        });
    }

    let resp = (StatusCode::ACCEPTED, Json(BundleAccepted { results })).into_response();
    Ok(with_quota_headers(resp, quota))
}

/// SPDX identifier of the licence a contributed manifest is placed under (§5b).
pub const CONTRIBUTION_LICENSE: &str = "CC0-1.0";

/// The grant a contributor makes, in the words §5b specifies.
///
/// Scope is the operative part: it covers *the manifest*, and cannot purport to
/// license the underlying work, which is not the contributor's to license and
/// which this server does not hold.
pub const CONTRIBUTION_TERMS: &str = "Contributing a manifest places its content \
    — timings, identifiers and audio signature — under CC0 1.0 Universal. This \
    covers the manifest only. It does not, and cannot, license the underlying \
    work, which the contributor does not own and this server does not hold.";

#[derive(Debug, Serialize)]
pub struct TokenIssued {
    pub token: String,
    /// Delivered with the capability, not merely published: a licence the server
    /// declares unilaterally is not one any contributor granted (§5b).
    pub contribution_license: &'static str,
    pub contribution_terms: &'static str,
}

/// Issues an anonymous bearer capability (§5a).
///
/// Self-issued on request: no email, no verification, no personal data. Stored
/// only as a hash, so the server cannot enumerate who holds tokens. Discarding a
/// token and requesting another is trivially easy — and that is fine, because the
/// token is not the defence; the content checks are.
///
/// The response carries the §5b contribution licence. This is the only moment
/// the server can obtain a grant: there are no accounts, so there is no sign-up
/// to attach terms to, and a manifest arrives with no channel to negotiate over.
/// Acquiring the capability is therefore where the grant has to be made.
/// TRACES: UR-005, UR-019 | SR-004 | PR-006
pub async fn post_token(
    State(state): State<AppState>,
    peer: crate::state::PeerIp,
    headers: HeaderMap,
) -> ApiResult<Json<TokenIssued>> {
    let ip = state.client_ip(&headers, peer.0);
    // Reuse the report budget: issuing tokens is cheap but should not be a free
    // unbounded write.
    state.check_limit(&ip, Surface::Report)?;

    let token = crate::auth::generate_token();
    let hash = crate::auth::hash_token(&token);
    let now = now_iso();
    state
        .db
        .write(move |tx| repo::insert_contributor(tx, &hash, &now))
        .await
        .map_err(ApiError::Internal)?;

    Ok(Json(TokenIssued {
        token,
        contribution_license: CONTRIBUTION_LICENSE,
        contribution_terms: CONTRIBUTION_TERMS,
    }))
}