//! UR-009 end to end — the audio signature through the whole server.
//!
//! The unit tests own the slide's arithmetic and the validator owns the
//! signature's shape. What only holds if the layers are composed correctly is
//! everything here: that a contributed signature survives storage and comes back
//! byte-identically, that the `audio` tier actually reaches the read endpoints,
//! and that a caller with no title identity at all can still find its film.
//!
//! **The round trip is the load-bearing one.** The signature exists so a client
//! can align a manifest against its own copy (`jRay` JR-047), and the client can
//! only do that if the manifest it fetches carries one. A server that validates
//! a signature and then drops it passes every validation test ever written and
//! delivers nothing.

use std::sync::Arc;

use axum::body::Body;
use axum::http::{Request, StatusCode};
use http_body_util::BodyExt;
use jray_server::db::{repo, Db};
use jray_server::ratelimit::RateLimiter;
use jray_server::state::AppState;
use jray_server::tmdb::TmdbClient;
use jray_server::{app, audio_sig, config::Config, worker};
use serde_json::{json, Value};
use tower::ServiceExt;

// ---------------------------------------------------------------------------
// Signatures under test
// ---------------------------------------------------------------------------

/// A real signature's frame count: a 120 s window at a 1024-sample hop (§3).
const FRAMES: usize = 1288;

/// A peak-bin sequence with the statistics film audio produces — no long runs,
/// no short period — packed into a wire signature.
///
/// Not the golden vector, which pins the *format* and is a tone staircase; an
/// alignment search over it is ambiguous in ways no film is. The offset study
/// that does use real audio is `scene-actor-extraction` VR-014.
fn signature(seed: u64, skip_frames: usize) -> String {
    let mut x = seed;
    let bytes: Vec<u8> = (0..FRAMES + skip_frames)
        .map(|_| {
            x = x.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
            // A whole byte: 5-bit band and 2-bit class, high bit clear.
            (((x >> 33) % 32) as u8) << 2 | ((x >> 29) % 4) as u8
        })
        .skip(skip_frames)
        .collect();
    audio_sig::encode(&bytes)
}

// ---------------------------------------------------------------------------
// Harness
// ---------------------------------------------------------------------------

struct TempDir(std::path::PathBuf);

impl TempDir {
    fn new(tag: &str) -> Self {
        let mut p = std::env::temp_dir();
        p.push(format!("jray-audio-{}-{}", tag, unique()));
        std::fs::create_dir_all(&p).expect("creating temp dir");
        Self(p)
    }
    fn db_path(&self) -> String {
        self.0.join("test.db").to_string_lossy().into_owned()
    }
}

impl Drop for TempDir {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}

fn unique() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static N: AtomicU64 = AtomicU64::new(0);
    let t = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!("{t}-{}", N.fetch_add(1, Ordering::Relaxed))
}

struct TestServer {
    router: axum::Router,
    db: Db,
    _dir: TempDir,
}

impl TestServer {
    fn new(tag: &str) -> Self {
        Self::with_search(tag, true)
    }

    fn with_search(tag: &str, enable_audio_search: bool) -> Self {
        let dir = TempDir::new(tag);
        let db = Db::open(&dir.db_path()).expect("opening database");
        let config = Arc::new(Config {
            bind: "127.0.0.1:0".into(),
            db_path: dir.db_path(),
            // Unconfigured, so uploads stay `pending` and no test reaches the
            // network. Anything that must be *served* is seeded as listed.
            tmdb_api_key: None,
            tmdb_base_url: "http://127.0.0.1:1".into(),
            trusted_proxies: Vec::new(),
            server_id: "audio.example".into(),
            publish_peer_directory: false,
            contact: None,
            enable_audio_search,
            request_timeout: std::time::Duration::from_secs(30),
            job_batch: 8,
            job_poll_interval: std::time::Duration::from_secs(3600),
        });
        let state = AppState {
            db: db.clone(),
            config: config.clone(),
            limiter: Arc::new(RateLimiter::new()),
            tmdb: Arc::new(TmdbClient::new(config.tmdb_base_url.clone(), None)),
        };
        Self { router: app::router(state), db, _dir: dir }
    }

    async fn send(&self, req: Request<Body>) -> (StatusCode, Value) {
        let resp = self.router.clone().oneshot(req).await.expect("router call");
        let status = resp.status();
        let bytes = resp.into_body().collect().await.expect("body").to_bytes();
        let body = if bytes.is_empty() {
            Value::Null
        } else {
            serde_json::from_slice(&bytes)
                .unwrap_or(Value::String(String::from_utf8_lossy(&bytes).into_owned()))
        };
        (status, body)
    }

    async fn get(&self, uri: &str) -> (StatusCode, Value) {
        self.send(Request::builder().uri(uri).body(Body::empty()).unwrap()).await
    }

    async fn post(&self, uri: &str, body: &Value) -> (StatusCode, Value) {
        self.send(
            Request::builder()
                .method("POST")
                .uri(uri)
                .header("content-type", "application/json")
                .body(Body::from(body.to_string()))
                .unwrap(),
        )
        .await
    }

    async fn post_auth(&self, uri: &str, token: &str, body: &Value) -> (StatusCode, Value) {
        self.send(
            Request::builder()
                .method("POST")
                .uri(uri)
                .header("content-type", "application/json")
                .header("authorization", format!("Bearer {token}"))
                .body(Body::from(body.to_string()))
                .unwrap(),
        )
        .await
    }

    async fn token(&self) -> String {
        let (status, body) = self.post("/api/v1/tokens", &json!({})).await;
        assert_eq!(status, StatusCode::OK, "token issue failed: {body}");
        body["token"].as_str().expect("token").to_string()
    }

    /// Seeds a listed manifest, since an upload stays `pending` without TMDB and
    /// `pending` is never served (§6, §7).
    async fn seed_listed(
        &self,
        tmdb_id: &str,
        title: &str,
        runtime_sec: f64,
        signature: Option<&str>,
    ) -> String {
        let tmdb_id = tmdb_id.to_string();
        let title = title.to_string();
        let packed = signature.and_then(audio_sig::decode);
        self.db
            .write(move |tx| {
                let now = worker::now_iso();
                let title_id = repo::upsert_title(
                    tx,
                    jray_server::model::IdentityType::Movie,
                    Some(&tmdb_id),
                    None,
                    Some(&title),
                    Some(1952),
                    &now,
                )?;
                let id = format!("m-{tmdb_id}");
                let coarse = packed.as_deref().map(audio_sig::coarse_key);
                repo::insert_manifest(
                    tx,
                    &repo::NewManifest {
                        id: &id,
                        title_id: &title_id,
                        season: None,
                        episode: None,
                        runtime_sec,
                        audio_signature: packed.as_deref(),
                        audio_sig_coarse: coarse.as_deref(),
                        sample_fps: Some(5.0),
                        extinction_sec: Some(12.0),
                        pipeline_version: Some("test 0.1"),
                        gallery_scope: Some("global"),
                        contributor_id: None,
                        status: "listed",
                        content_id: Some(&format!("cid-{tmdb_id}")),
                        origin: "audio.example",
                        ingested_from: None,
                        created_at: &now,
                    },
                )?;
                repo::upsert_person(tx, 884, "Bob Hope", false, &now)?;
                repo::insert_actor_scenes(
                    tx,
                    &id,
                    884,
                    &[jray_server::validate::SceneCs::plain(19160, 20920)],
                )?;
                Ok(id)
            })
            .await
            .expect("seeding")
    }
}

fn movie_manifest(tmdb_id: &str, runtime: f64, signature: Option<&str>) -> Value {
    let mut cut = json!({ "runtime_sec": runtime });
    if let Some(sig) = signature {
        cut["audio_signature"] = json!(sig);
    }
    json!({
        "jmanifest_version": 2,
        "identity": { "type": "movie", "tmdb_id": tmdb_id, "title": "Road to Bali", "year": 1952 },
        "cut": cut,
        "extraction": { "sample_fps": 5, "pipeline_version": "test 0.1" },
        "actors": [
            { "name": "Bob Hope", "tmdb_id": "884", "scenes": [{"start":191.6,"end":209.2}] },
            { "name": "Bing Crosby", "tmdb_id": "11007", "scenes": [{"start":300.0,"end":320.0}] }
        ]
    })
}

// ---------------------------------------------------------------------------
// Storage and the round trip
// ---------------------------------------------------------------------------

/// TRACES: UR-009 | DR-002 | SR-003
#[tokio::test]
async fn a_contributed_signature_survives_storage_byte_for_byte() {
    // The gap this closes: a signature that is validated and then dropped leaves
    // every fetched manifest without one, and the plugin's local alignment
    // (JR-047) with nothing to align against.
    let s = TestServer::new("roundtrip");
    let token = s.token().await;
    let sig = signature(11, 0);

    let (status, body) = s
        .post_auth("/api/v1/manifests", &token, &movie_manifest("504172", 6420.5, Some(&sig)))
        .await;
    assert_eq!(status, StatusCode::ACCEPTED, "{body}");
    let id = body["manifest_id"].as_str().expect("manifest id").to_string();

    // Listed by hand, since the cast check cannot run without TMDB.
    let listed = id.clone();
    s.db.write(move |tx| {
        tx.execute("UPDATE manifests SET status = 'listed' WHERE id = ?1", [&listed])?;
        Ok(())
    })
    .await
    .unwrap();

    let (status, body) = s.get(&format!("/api/v1/manifests/{id}")).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(
        body["cut"]["audio_signature"].as_str(),
        Some(sig.as_str()),
        "the served signature must be the contributed one, re-encoded not echoed"
    );
}

/// TRACES: UR-009 | SR-003
#[tokio::test]
async fn a_manifest_without_a_signature_serves_no_signature_field() {
    // §3's sequencing: signatures accumulate, so most manifests have none for a
    // long time. The field must be absent rather than null or empty.
    let s = TestServer::new("nosig");
    let id = s.seed_listed("100", "Unsigned", 6420.5, None).await;
    let (status, body) = s.get(&format!("/api/v1/manifests/{id}")).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body["cut"].get("audio_signature").is_none(), "{body}");
}

/// TRACES: UR-008, UR-009 | SR-003
#[tokio::test]
async fn a_held_manifest_adopts_a_signature_it_lacked() {
    // §9a: the signature is excluded from `content_id` and replicates as an
    // attribute, so the same content arriving with a signature must fill the
    // hole. Otherwise a manifest that first arrived without audio could never
    // gain one, and its cut would sit permanently outside the `audio` tier.
    let s = TestServer::new("adopt");
    let token = s.token().await;
    let sig = signature(12, 0);

    let (status, _) =
        s.post_auth("/api/v1/manifests", &token, &movie_manifest("777", 6420.5, None)).await;
    assert_eq!(status, StatusCode::ACCEPTED);

    // Same content, now carrying a signature. Deduplicated by `content_id` —
    // which the signature is deliberately not part of.
    let (status, body) =
        s.post_auth("/api/v1/manifests", &token, &movie_manifest("777", 6420.5, Some(&sig))).await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(body["status"], "already_present");
    let id = body["manifest_id"].as_str().unwrap().to_string();

    let listed = id.clone();
    s.db.write(move |tx| {
        tx.execute("UPDATE manifests SET status = 'listed' WHERE id = ?1", [&listed])?;
        Ok(())
    })
    .await
    .unwrap();

    let (_, body) = s.get(&format!("/api/v1/manifests/{id}")).await;
    assert_eq!(body["cut"]["audio_signature"].as_str(), Some(sig.as_str()));
}

// ---------------------------------------------------------------------------
// The `audio` tier on the read endpoints
// ---------------------------------------------------------------------------

/// TRACES: UR-009 | SR-003
#[tokio::test]
async fn a_matching_signature_reaches_the_audio_tier_on_a_fetch() {
    let s = TestServer::new("tier");
    let sig = signature(13, 0);
    s.seed_listed("504172", "Road to Bali", 6420.5, Some(&sig)).await;

    let uri = format!(
        "/api/v1/manifests/movie?tmdb_id=504172&runtime_sec=6420.5&audio_signature={}",
        urlencode(&sig)
    );
    let (status, body) = s.get(&uri).await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(body["match"], "audio");
    assert_eq!(body["offset_sec"], 0.0);
}

/// TRACES: UR-009 | SR-003
#[tokio::test]
async fn a_shifted_release_matches_and_gets_its_offset() {
    // The row §3 calls the valuable one: a release with extra head material
    // previously failed the ±2 s runtime tier outright. Now it matches, and the
    // client shifts every window by the recovered offset.
    let s = TestServer::new("shifted");
    let stored_runtime = 6420.5;
    let head_sec = 40.0;
    let sig = signature(14, 0);
    s.seed_listed("504172", "Road to Bali", stored_runtime, Some(&sig)).await;

    // The client's copy carries 40 s of extra logos: 40 s longer, and its
    // centre window therefore starts 20 s later in the content.
    let slide = ((head_sec / 2.0) / audio_sig::HOP_SEC).round() as usize;
    let client_sig = signature(14, slide);

    let uri = format!(
        "/api/v1/manifests/movie?tmdb_id=504172&runtime_sec={}&audio_signature={}",
        stored_runtime + head_sec,
        urlencode(&client_sig)
    );
    let (status, body) = s.get(&uri).await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(body["match"], "audio");

    // Both terms cancel here — the window moved by exactly the slide — so the
    // manifest's timings apply unshifted despite a 40 s runtime difference the
    // `runtime` tier would have rejected outright.
    let offset = body["offset_sec"].as_f64().unwrap();
    assert!(offset.abs() <= audio_sig::HOP_SEC, "offset {offset}");

    // And without the signature that same request is not a match at all.
    let (status, _) = s
        .get(&format!(
            "/api/v1/manifests/movie?tmdb_id=504172&runtime_sec={}",
            stored_runtime + head_sec
        ))
        .await;
    assert_eq!(status, StatusCode::NOT_FOUND);
}

/// TRACES: UR-009 | SR-003
#[tokio::test]
async fn a_signature_that_disagrees_is_not_served_on_a_runtime_coincidence() {
    // §3: `audio` outranks `runtime` because it is content-derived, so its
    // refusal outranks a runtime agreement. Two different films of the same
    // length must not match.
    let s = TestServer::new("disagree");
    s.seed_listed("504172", "Road to Bali", 6420.5, Some(&signature(15, 0))).await;

    let uri = format!(
        "/api/v1/manifests/movie?tmdb_id=504172&runtime_sec=6420.5&audio_signature={}",
        urlencode(&signature(16, 0))
    );
    let (status, _) = s.get(&uri).await;
    assert_eq!(status, StatusCode::NOT_FOUND);
}

/// TRACES: UR-001, UR-009 | SR-003
#[tokio::test]
async fn exists_reports_the_audio_tier_too() {
    // The sweep path. `exists` transfers no payload, so a client cannot align
    // locally from it — server-side matching is the only way it can say `audio`.
    let s = TestServer::new("exists");
    let sig = signature(17, 0);
    s.seed_listed("504172", "Road to Bali", 6420.5, Some(&sig)).await;

    let (status, body) = s
        .post(
            "/api/v1/manifests/exists",
            &json!({ "items": [
                { "tmdb_id": "504172", "runtime_sec": 6420.5, "audio_signature": sig },
                { "tmdb_id": "504172", "runtime_sec": 6420.5 }
            ]}),
        )
        .await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(body["results"][0]["match"], "audio");
    assert_eq!(body["results"][1]["match"], "runtime");
}

/// TRACES: UR-009 | DR-013 | SR-003
#[tokio::test]
async fn a_signature_without_a_runtime_is_refused_by_name() {
    // The offset's window-anchor term is derived from both runtimes (§3). Rather
    // than answer at a lower tier — which would look like a match the client's
    // own signature had failed to improve — the request is refused.
    let s = TestServer::new("noruntime");
    s.seed_listed("504172", "Road to Bali", 6420.5, Some(&signature(18, 0))).await;

    let uri = format!(
        "/api/v1/manifests/movie?tmdb_id=504172&audio_signature={}",
        urlencode(&signature(18, 0))
    );
    let (status, body) = s.get(&uri).await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
    assert!(body.to_string().contains("runtime_sec"), "{body}");
}

/// TRACES: UR-009, UR-011 | DR-013 | SR-004
#[tokio::test]
async fn a_malformed_signature_is_a_bad_request_not_a_silent_downgrade() {
    let s = TestServer::new("malformed");
    s.seed_listed("504172", "Road to Bali", 6420.5, Some(&signature(19, 0))).await;

    for bad in ["v2:AAAA", "notasignature", "v1:!!!!"] {
        let uri = format!(
            "/api/v1/manifests/movie?tmdb_id=504172&runtime_sec=6420.5&audio_signature={}",
            urlencode(bad)
        );
        let (status, _) = s.get(&uri).await;
        assert_eq!(status, StatusCode::BAD_REQUEST, "accepted {bad}");
    }
}

// ---------------------------------------------------------------------------
// Unknown-providence search
// ---------------------------------------------------------------------------

/// TRACES: UR-009 | SR-003 | PR-005
#[tokio::test]
async fn search_identifies_a_file_with_no_metadata_at_all() {
    // What the endpoint is for: no TMDB id, no usable name, nothing to look up.
    // The answer has to carry title identity, because the caller has none.
    let s = TestServer::new("search");
    let sig = signature(20, 0);
    s.seed_listed("504172", "Road to Bali", 6420.5, Some(&sig)).await;
    s.seed_listed("999", "A Different Film", 6420.0, Some(&signature(21, 0))).await;

    let (status, body) = s
        .post("/api/v1/manifests/search", &json!({ "audio_signature": sig, "runtime_sec": 6420.5 }))
        .await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(body["results"].as_array().unwrap().len(), 1, "{body}");
    assert_eq!(body["results"][0]["identity"]["tmdb_id"], "504172");
    assert_eq!(body["results"][0]["identity"]["title"], "Road to Bali");
    assert_eq!(body["results"][0]["match"], "audio");
    assert_eq!(body["results"][0]["score"], 1.0);
    assert_eq!(body["truncated"], false);
    // The decoy shares a runtime and so cleared the prefilter — the slide is
    // what rejected it, which is the whole point of scoring rather than
    // shortlisting.
    assert_eq!(body["candidates_scored"], 2);
}

/// TRACES: UR-009 | SR-003
#[tokio::test]
async fn search_recovers_the_offset_for_a_differently_trimmed_release() {
    let s = TestServer::new("search-offset");
    let stored_runtime = 6420.5;
    s.seed_listed("504172", "Road to Bali", stored_runtime, Some(&signature(22, 0))).await;

    // 30 s of extra head: the window moves 15 s later in the content.
    let head_sec = 30.0;
    let slide = ((head_sec / 2.0) / audio_sig::HOP_SEC).round() as usize;

    let (status, body) = s
        .post(
            "/api/v1/manifests/search",
            &json!({
                "audio_signature": signature(22, slide),
                "runtime_sec": stored_runtime + head_sec
            }),
        )
        .await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(body["results"][0]["identity"]["tmdb_id"], "504172");
    let offset = body["results"][0]["offset_sec"].as_f64().unwrap();
    assert!(offset.abs() <= audio_sig::HOP_SEC, "offset {offset}");
}

/// TRACES: UR-009 | SR-003
#[tokio::test]
async fn search_declines_content_it_does_not_hold() {
    let s = TestServer::new("search-miss");
    s.seed_listed("504172", "Road to Bali", 6420.5, Some(&signature(23, 0))).await;

    let (status, body) = s
        .post(
            "/api/v1/manifests/search",
            &json!({ "audio_signature": signature(24, 0), "runtime_sec": 6420.5 }),
        )
        .await;
    assert_eq!(status, StatusCode::OK);
    assert!(body["results"].as_array().unwrap().is_empty(), "{body}");
    // An empty result is an answer, not an error: the caller learns the
    // community does not have this cut.
    assert_eq!(body["candidates_scored"], 1);
}

/// TRACES: UR-009, UR-011 | SR-004
#[tokio::test]
async fn search_applies_the_same_structural_rules_as_an_upload() {
    // §6: a read must not be a way to hand the server bytes an upload would
    // have refused.
    let s = TestServer::new("search-validate");
    for (body, why) in [
        (json!({ "audio_signature": "v1:AAAA", "runtime_sec": 6420.5 }), "wrong length"),
        (json!({ "audio_signature": signature(25, 0), "runtime_sec": 60.0 }), "under the window"),
        (json!({ "audio_signature": signature(25, 0), "runtime_sec": -1.0 }), "negative runtime"),
    ] {
        let (status, _) = s.post("/api/v1/manifests/search", &body).await;
        assert_eq!(status, StatusCode::BAD_REQUEST, "accepted a signature that is {why}");
    }
}

/// TRACES: UR-009, UR-014 | SR-003
#[tokio::test]
async fn search_is_absent_when_the_operator_has_not_enabled_it() {
    // §3: optional for a server to implement, and advertised rather than
    // assumed, so a client discovers the absence in one cheap request.
    let s = TestServer::with_search("search-off", false);
    let (status, body) = s.get("/api/v1/federation/capabilities").await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["audio_search"], false);
    // Matching is not the expensive half and stays on regardless.
    assert_eq!(body["audio_tier_matching"], true);

    let (status, _) = s
        .post(
            "/api/v1/manifests/search",
            &json!({ "audio_signature": signature(26, 0), "runtime_sec": 6420.5 }),
        )
        .await;
    assert_eq!(status, StatusCode::NOT_FOUND);
}

/// TRACES: UR-002, UR-009 | SR-005
#[tokio::test]
async fn search_never_returns_a_pending_manifest() {
    // §6/§7: `pending` is held unlisted and is not served by any route. A search
    // that leaked one would publish an unchecked contribution.
    let s = TestServer::new("search-pending");
    let token = s.token().await;
    let sig = signature(27, 0);
    let (status, _) = s
        .post_auth("/api/v1/manifests", &token, &movie_manifest("504172", 6420.5, Some(&sig)))
        .await;
    assert_eq!(status, StatusCode::ACCEPTED);

    let (status, body) = s
        .post("/api/v1/manifests/search", &json!({ "audio_signature": sig, "runtime_sec": 6420.5 }))
        .await;
    assert_eq!(status, StatusCode::OK);
    assert!(body["results"].as_array().unwrap().is_empty(), "{body}");
    assert_eq!(body["candidates_scored"], 0);
}

/// Percent-encodes the base64 characters a query string would otherwise eat.
fn urlencode(s: &str) -> String {
    s.chars()
        .map(|c| match c {
            '+' => "%2B".to_string(),
            '/' => "%2F".to_string(),
            '=' => "%3D".to_string(),
            ':' => "%3A".to_string(),
            '!' => "%21".to_string(),
            c => c.to_string(),
        })
        .collect()
}