//! Admission by age at the room service (WAI age-assurance §5.6), with a relay on loopback
//! that trusts its authority:
//!
//! - a viewer's grant request without a token is answered `401` with a `PrivateToken`
//!   challenge for each key of the room's policy (RFC 9577 §2.1.2), each naming the service's
//!   server name and a window's redemption context;
//! - a token of a key at or above the room's age is admitted, and the grant states its
//!   threshold in `age` (§5.4), which the relay reads; one below is refused `below_threshold`,
//!   one spent twice `token_replayed`, one for another origin's challenge
//!   `token_wrong_challenge`, and one that does not read `token_malformed`;
//! - a publisher is challenged as a viewer is, and enters at 18; the room's rating and its own
//!   rating of its video at 18, carried on its PUBLISHes (§5.7), have the relay serve the viewer
//!   at 16 the audio and refuse it the video (`below_threshold`), and serve the viewer at 18
//!   both;
//! - the admission receipt (§7) reads and checks under the room's policy, every count of it
//!   under 50 suppressed.

#[path = "common/rfc_test_keys.rs"]
mod rfc_test_keys;

use ed25519_dalek::SigningKey;
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
use joule_moq::draft22::Parameters;
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, Grant};
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::websocket;
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_room::age::{b64url, unb64url};
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, Config, Service};
use rfc_test_keys::{k13, k16, k18, mint};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;

const STEP: Duration = Duration::from_secs(20);

async fn relay(keys: &Keys) -> RelayServer {
    let mut c = RelayConfig::loopback();
    c.quic_listen = None;
    c.auth.required = true;
    c.auth.authorities = vec![Authority { key: hex::encode(keys.authority.verifying_key().to_bytes()), rooms: vec![] }];
    let t = hex::encode(Sha256::digest(keys.clock_token));
    let clock = hex::encode(keys.clock.verifying_key().to_bytes());
    c.auth.grants = vec![
        Grant { prefix: vec!["*".into(), clock, "events".into()], token_sha256: t.clone(), publish: true, subscribe: false, proof: false, unlimited: true, seek: false },
        Grant { prefix: vec![], token_sha256: t, publish: false, subscribe: true, proof: false, unlimited: false, seek: false },
    ];
    RelayServer::start(c).await.expect("the relay starts")
}

/// One HTTP/1.1 request: its status, its header fields (names in lowercase) and its JSON.
async fn call(addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>, client: &str, authorization: Option<&str>) -> (u16, Vec<(String, String)>, Value) {
    let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
    let b = body.map(|v| v.to_string()).unwrap_or_default();
    let auth = authorization.map(|a| format!("Authorization: {a}\r\n")).unwrap_or_default();
    let req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\nCF-Connecting-IP: {client}\r\n{auth}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}", b.len());
    s.write_all(req.as_bytes()).await.unwrap();
    let mut out = Vec::new();
    timeout(STEP, s.read_to_end(&mut out)).await.expect("an answer in time").unwrap();
    let text = String::from_utf8(out).unwrap();
    let (head, body) = text.split_once("\r\n\r\n").expect("a response");
    let mut lines = head.split("\r\n");
    let status: u16 = lines.next().unwrap().split(' ').nth(1).unwrap().parse().unwrap();
    let headers = lines.filter_map(|l| l.split_once(':')).map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())).collect();
    (status, headers, serde_json::from_str(body).unwrap_or(Value::Null))
}

fn key(i: u8) -> SigningKey {
    SigningKey::from_bytes(&[i; 32])
}

fn pubhex(k: &SigningKey) -> String {
    hex::encode(k.verifying_key().to_bytes())
}

async fn ask(addr: std::net::SocketAddr, k: &SigningKey, role: &str, client: &str, token: Option<&[u8]>) -> (u16, Vec<(String, String)>, Value) {
    let auth = token.map(|t| format!("PrivateToken token=\"{}\"", b64url(t)));
    call(addr, "POST", "/grant", Some(&json!({"principal": pubhex(k), "role": role})), client, auth.as_deref()).await
}

async fn session(server: &RelayServer, grant: &[u8], k: &SigningKey) -> Client {
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let dial = || async { Ok::<joule_moq_relay::transport::SharedConnection, String>(Arc::new(websocket::connect(&url, MuxConfig::default()).await.map_err(|e| e.to_string())?)) };
    let opts = ClientOptions { grants: vec![(grant.to_vec(), k.clone())], target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path), ..ClientOptions::default() };
    // The relay reads this SETUP before it sends its own, so the grant is proved over a
    // challenge its SETUP gave: on a second session, where the first held none.
    Client::connect_proving(dial, opts).await.expect("SETUP")
}

/// A `WWW-Authenticate: PrivateToken` challenge's TokenChallenge and key, read.
fn challenge_of(v: &str) -> (Vec<u8>, Vec<u8>) {
    let rest = v.strip_prefix("PrivateToken ").expect("the scheme");
    let p = |n: &str| rest.split(',').map(str::trim).find_map(|x| x.strip_prefix(n)).map(|x| unb64url(x.trim_matches('"')).expect("base64url")).expect(n);
    (p("challenge="), p("token-key="))
}

fn age_of(grant_hex: &str, keys: &Keys) -> u8 {
    joule_moq::grant_proof::GrantHead::verified(&hex::decode(grant_hex).unwrap(), &keys.authority.verifying_key().to_bytes()).expect("a grant the authority signed").age
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_room_admits_viewers_by_a_privacy_pass_token_and_the_relay_serves_by_its_threshold() {
    timeout(Duration::from_secs(240), run()).await.expect("the test finishes in time");
}

async fn run() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let (k13, k16, k18) = (k13(), k16(), k18());
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\nmax_publishers = 2\njoin_wait_s = 30\n[rate]\nclient_burst = 40\nclient_every_s = 1\n\
         [age]\ntoken_window_s = 600\n\
         [[age.issuers]]\nname = \"issuer-a.example\"\nkeys = [{{ threshold = 13, spki = \"{}\" }}, {{ threshold = 16, spki = \"{}\" }}]\n\
         [[age.issuers]]\nname = \"issuer-b.example\"\nkeys = [{{ threshold = 18, spki = \"{}\" }}]\n\
         [[age.ratings]]\nscheme = \"https://rating-a.example/scheme\"\nvalue = \"16\"\nmin_age = 16\n",
        server.websocket_addr.unwrap(),
        server.relay().config().path,
        hex::encode(&k13.spki),
        hex::encode(&k16.spki),
        hex::encode(&k18.spki),
    ))
    .expect("the configuration reads");
    let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
    let listener = tokio::net::TcpListener::bind(cfg.listen).await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(svc.clone().run_clock());
    tokio::spawn(http::serve(svc.clone(), listener));
    assert!(timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .is_ok());

    // A publisher is challenged too, and enters with a token of 18.
    let alice = key(0x41);
    let (st, h, body) = ask(addr, &alice, "publisher", "10.1.0.1", None).await;
    assert_eq!((st, body["error"].as_str()), (401, Some("token_required")), "{body}");
    let ch18 = h.iter().filter(|(k, _)| k == "www-authenticate").map(|(_, v)| challenge_of(v)).find(|(_, k)| *k == k18.spki).expect("a challenge of the 18 key").0;
    let (st, _, a) = ask(addr, &alice, "publisher", "10.1.0.1", Some(&mint(&k18, &ch18, 9))).await;
    assert_eq!(st, 200, "{a}");
    assert_eq!(age_of(a["grant"].as_str().unwrap(), &keys), 18);
    let room = a["room"]["room"].as_str().unwrap().to_string();
    let age = &a["room"]["age"];
    assert_eq!(age["age"].as_u64(), Some(16), "{age}");
    let ratings: Vec<String> = age["ratings"].as_array().unwrap().iter().map(|r| r.as_str().unwrap().to_string()).collect();
    let descriptor = wai::live_room::Descriptor::parse(&hex::decode(a["room"]["descriptor"].as_str().unwrap()).unwrap()).unwrap();
    let policy = wai::age_assurance::read_policy(age["policy"].as_str().unwrap(), &descriptor).expect("the room's policy reads");
    let claim = wai::age_assurance::read_claim(&ratings[0]).expect("the room's rating reads");
    assert_eq!((claim.age(), claim.signer), (16, keys.authority.verifying_key().to_bytes()));

    // A viewer without a token: challenged, once for each key of the policy.
    let (st, h, body) = ask(addr, &key(0x42), "viewer", "10.1.0.2", None).await;
    assert_eq!((st, body["error"].as_str(), body["age"].as_u64()), (401, Some("token_required"), Some(16)), "{body}");
    let offered: Vec<(Vec<u8>, Vec<u8>)> = h.iter().filter(|(k, _)| k == "www-authenticate").map(|(_, v)| challenge_of(v)).collect();
    assert_eq!(offered.len(), 3, "{h:?}");
    assert_eq!(offered.iter().map(|(_, k)| k.clone()).collect::<Vec<_>>(), vec![k13.spki.clone(), k16.spki.clone(), k18.spki.clone()], "in ascending order of threshold");
    for (c, _) in &offered {
        let c = wai::age_assurance::read_challenge(c).expect("RFC 9577's TokenChallenge");
        assert_eq!((c.token_type, c.origin_info.as_slice(), c.redemption_context.len()), (2, &b"room.test"[..], 32));
    }
    let ch = |i: usize| offered[i].0.clone();

    // Below the room's age: refused, the token spent.
    let (st, h, body) = ask(addr, &key(0x43), "viewer", "10.1.0.3", Some(&mint(&k13, &ch(0), 1))).await;
    assert_eq!((st, body["error"].as_str()), (403, Some("below_threshold")), "{body}");
    assert_eq!(h.iter().filter(|(k, _)| k == "www-authenticate").count(), 3, "challenged again");
    // At 16: admitted, and the grant states 16.
    let t16 = mint(&k16, &ch(1), 2);
    let v16 = key(0x44);
    let (st, _, g16) = ask(addr, &v16, "viewer", "10.1.0.4", Some(&t16)).await;
    assert_eq!(st, 200, "{g16}");
    assert_eq!(age_of(g16["grant"].as_str().unwrap(), &keys), 16);
    // The same token again: spent.
    let (st, _, body) = ask(addr, &key(0x45), "viewer", "10.1.0.5", Some(&t16)).await;
    assert_eq!((st, body["error"].as_str()), (403, Some("token_replayed")), "{body}");
    // At 18, by the second issuer.
    let (st, _, g18) = ask(addr, &key(0x46), "viewer", "10.1.0.6", Some(&mint(&k18, &ch(2), 3))).await;
    assert_eq!((st, age_of(g18["grant"].as_str().unwrap(), &keys)), (200, 18));
    // A token for another origin's challenge, and one that does not read.
    let other = wai::age_assurance::challenge_bytes(2, b"issuer-a.example", &wai::age_assurance::read_challenge(&ch(1)).unwrap().redemption_context, b"other.test");
    let (st, _, body) = ask(addr, &key(0x47), "viewer", "10.1.0.7", Some(&mint(&k16, &other, 4))).await;
    assert_eq!((st, body["error"].as_str()), (403, Some("token_wrong_challenge")), "{body}");
    let (st, _, body) = call(addr, "POST", "/grant", Some(&json!({"principal": pubhex(&key(0x48)), "role": "viewer"})), "10.1.0.8", Some("PrivateToken token=\"***\"")).await;
    assert_eq!((st, body["error"].as_str()), (403, Some("token_malformed")), "{body}");

    // The relay: the room's rating rides alice's PUBLISHes, and her own rating of her video at
    // 18 rides its PUBLISH; the viewer at 16 is served her audio and refused her video, and the
    // viewer at 18 is served both.
    let ca = session(&server, &hex::decode(a["grant"].as_str().unwrap()).unwrap(), &alice).await;
    let track = |m: &str| TrackNamespace::from_strs([room.as_str(), pubhex(&alice).as_str(), m]);
    let mut held = Vec::new();
    for (i, m) in ["audio", "video"].iter().enumerate() {
        let mut props = Properties::new();
        props.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 64));
        for r in &ratings {
            props.push(joule_moq::rating::property(r));
        }
        if *m == "video" {
            let digest = joule_moq::rating::track_digest(&track(m), &TrackName::from_str("0"));
            let own = joule_moq::rating::write(&alice, &joule_moq::rating::Subject::Track { track: digest, from_group: 0 }, &[joule_moq::rating::Rating { scheme: "https://rating-a.example/scheme".into(), value: "18".into(), min_age: 18 }], "2026-10-21T00:00:00Z");
            props.push(joule_moq::rating::property(&own));
        }
        held.push(ca.publish(track(m), TrackName::from_str("0"), 10 + i as u64, Parameters::new(), props).await.expect("PUBLISH"));
    }
    assert_eq!(server.relay().stats().ratings_held, 2, "the room's claim, and alice's about her video");
    let cv = session(&server, &hex::decode(g16["grant"].as_str().unwrap()).unwrap(), &v16).await;
    cv.subscribe(track("audio"), TrackName::from_str("0"), Parameters::new()).await.expect("the viewer at 16 is served the audio");
    let refused = cv.subscribe(track("video"), TrackName::from_str("0"), Parameters::new()).await.expect_err("the video is rated 18");
    assert!(refused.contains("below_threshold"), "{refused}");
    let v18 = key(0x4a);
    let (st, _, g) = ask(addr, &v18, "viewer", "10.1.0.9", Some(&mint(&k18, &ch(2), 10))).await;
    assert_eq!(st, 200, "{g}");
    let c18 = session(&server, &hex::decode(g["grant"].as_str().unwrap()).unwrap(), &v18).await;
    c18.subscribe(track("video"), TrackName::from_str("0"), Parameters::new()).await.expect("the viewer at 18 is served the video");

    // The receipt: four admissions, three at 18 by one issuer and one at 16 by the other, each
    // cell under 50, so suppressed.
    let (st, _, r) = call(addr, "GET", "/age/receipt", None, "10.1.0.10", None).await;
    assert_eq!(st, 200, "{r}");
    let rc = wai::age_assurance::read_receipt(r["receipt"].as_str().unwrap(), &descriptor, &policy).expect("the receipt reads and checks");
    let cells: Vec<(String, u64, u64, bool)> = rc.cells.iter().map(|c| (c.issuer.clone(), c.threshold, c.admitted, c.suppressed)).collect();
    assert_eq!(cells, vec![("issuer-a.example".into(), 16, 0, true), ("issuer-b.example".into(), 18, 0, true)]);
    drop(held);
    for c in [&ca, &cv, &c18] {
        c.close(0, "done");
    }
}