//! The room's own service under the serving rule (WAI age-assurance §5.5), in a room that
//! admits by age and keeps co-performance: its tracks rated 18 by the room's authority, and
//! the relay holding the room's floor, its clock a room key (`[floor] room_keys`).
//!
//! The room's clock follows and logs every performer's objects over a session of its own. A
//! session that proves a room key in its SETUP (a GRANT_PROOF naming no grant, over the
//! session's challenge) is the room's service, and the relay serves it the rated tracks:
//!
//! - a session proving the clock's key, and one proving the authority's, receive every group;
//! - the same session with the clock's token and no proof, and one proving a key that is no
//!   room key, are ordinary sessions at threshold 0: refused the subscription
//!   (`below_threshold`);
//! - a proof that names the clock's key under another key's signature, and one by the clock
//!   over another session's challenge, close the session with UNAUTHORIZED.
//!
//! Over the WebSocket binding.

mod common;

use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::draft22::kvp::KeyValuePair;
use joule_moq::draft22::{Parameters, RequestErrorCode, SessionError, Token};
use joule_moq::grant_proof::{self, GrantProof, NO_GRANT, SETUP_OPTION_GRANT_PROOF};
use joule_moq::rating::{self, Rating, Subject};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, FloorConfig, Grant};
use joule_moq_relay::transport::TransportError;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time::timeout;

const AT: &str = "2026-10-22T00:00:00Z";

fn now_s() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
}

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

/// A grant laid out as live-rooms §4.1 lays it out, with the seek power, `publish` the media
/// bits, and the attributes `attrs` (§4.4), signed by `authority`.
fn grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, attrs: &[(&str, &str)]) -> Vec<u8> {
    let now = now_s();
    let mut g = b"WRG1\x01".to_vec();
    g.extend_from_slice(&room);
    g.extend_from_slice(&principal.verifying_key().to_bytes());
    g.extend_from_slice(&[0; 40]);
    g.extend_from_slice(&(now - 60).to_le_bytes());
    g.extend_from_slice(&(now + 600).to_le_bytes());
    g.extend_from_slice(&[0, publish, 1, 0, 0x10, 0]);
    g.extend_from_slice(&[0; 32]);
    g.extend_from_slice(b"WEP1\x01");
    g.extend_from_slice(&[0; 32]);
    g.extend_from_slice(&[0; 28]);
    g.extend_from_slice(&[0xff; 16]);
    g.extend_from_slice(&[0; 3]);
    let a: Vec<u8> = attrs.iter().flat_map(|(k, v)| [&[k.len() as u8][..], k.as_bytes(), &(v.len() as u16).to_le_bytes(), v.as_bytes()].concat()).collect();
    g.extend_from_slice(&[0, 0]);
    g.extend_from_slice(&(a.len() as u16).to_le_bytes());
    g.extend_from_slice(&a);
    let mut m = grant_proof::GRANT_DOMAIN.to_vec();
    m.extend_from_slice(&g);
    g.extend_from_slice(&authority.sign(&m).to_bytes());
    g
}

fn media_groups(got: &[Got], alias: u64) -> Vec<u64> {
    let mut g: Vec<u64> = got.iter().filter(|x| x.header.track_alias == alias && x.reset.is_none() && !x.objects.is_empty()).map(|x| x.header.group_id).collect();
    g.sort();
    g.dedup();
    g
}

async fn closed_unauthorized(c: &Client, what: &str) {
    match timeout(STEP, c.conn.closed()).await.expect("closed in time") {
        TransportError::Closed { code, .. } => assert_eq!(code, SessionError::UNAUTHORIZED.0, "{what}"),
        other => panic!("{what}: {other:?}"),
    }
}

async fn below_threshold(c: &Client, t: &Tracks, what: &str) {
    let refused = c.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.expect_err(what);
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()) && refused.contains("below_threshold"), "{what}: {refused}");
}

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

async fn run() {
    let authority = SigningKey::from_bytes(&[0x81; 32]);
    let clock = SigningKey::from_bytes(&[0x82; 32]);
    let stranger = SigningKey::from_bytes(&[0x83; 32]);
    let performer = SigningKey::from_bytes(&[0x84; 32]);
    let room = [0x4d; 32];
    let r = hex::encode(room);
    let token = b"the room clock's token".to_vec();
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
        // The clock's own rule: it subscribes to every track, by its token; exempt from limits.
        c.auth.grants = vec![Grant { prefix: vec![], token_sha256: hex::encode(joule_moq_relay::auth::sha256(&token)), publish: false, subscribe: true, proof: false, unlimited: true, seek: false }];
        // Co-performance: the relay holds the room's floor, the clock a room key.
        c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() });
    })
    .await;
    let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let t = Tracks::new(ns(&[&r, &hexkey(&performer), "video"]), "0", 10, 5, 128);

    // The performer, admitted at 18, publishes; its PUBLISH carries the room's rating, 18.
    let p_opts = ClientOptions { grants: vec![(grant(&authority, room, &performer, 0b011, &[("age", "18")]), performer.clone())], target: target.clone(), ..ClientOptions::default() };
    let (p, _) = connect(&server, Via::WebSocket, p_opts).await;
    let mut props = track_props(t.priority, None);
    let rated = vec![Rating { scheme: "https://rating-a.example/scheme".into(), value: "18".into(), min_age: 18 }];
    props.push(rating::property(&rating::write(&authority, &Subject::Room(room), &rated, AT)));
    let _media = p.publish(t.ns.clone(), t.media.clone(), t.media_alias, Parameters::new(), props).await.expect("PUBLISH the media track");
    let _receipts = p.publish(t.ns.clone(), t.receipts.clone(), t.receipts_alias, Parameters::new(), track_props(t.priority, None)).await.expect("PUBLISH the receipts track");
    assert_eq!(server.relay().stats().ratings_held, 1, "the authority's rating of the room is held");

    let session = |room_keys: Vec<SigningKey>| ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: token.clone() }], room_keys, target: target.clone(), ..ClientOptions::default() };

    // The clock's token with no proof: an ordinary session, at 0, refused the room's tracks.
    let (ordinary, _) = connect(&server, Via::WebSocket, session(vec![])).await;
    below_threshold(&ordinary, &t, "the clock's token alone").await;
    // A proof that passes by a key that is no room key: an ordinary session too.
    let (other, _) = connect(&server, Via::WebSocket, session(vec![stranger.clone()])).await;
    below_threshold(&other, &t, "a proof by a key that is no room key").await;

    // Proofs of the clock's key, and of the authority's: the room's service, served.
    let (by_clock, _) = connect(&server, Via::WebSocket, session(vec![clock.clone()])).await;
    let (by_authority, _) = connect(&server, Via::WebSocket, session(vec![authority.clone()])).await;
    let s_clock = subscribe_both(&by_clock, &t, Parameters::new()).await;
    let s_authority = subscribe_both(&by_authority, &t, Parameters::new()).await;
    for g in 0..5 {
        publish_group(&p, &performer, &t, g, &payload("p", g, 200)).await;
    }
    let (got_clock, got_authority) = tokio::join!(
        // Until every group has arrived, each on a stream of its own and in any order.
        collect_concurrent(&by_clock, STEP, Duration::from_millis(700), |g| { let m = media_groups(g, s_clock.media_alias); (0..5).all(|n| m.contains(&n)) }),
        collect_concurrent(&by_authority, STEP, Duration::from_millis(700), |g| { let m = media_groups(g, s_authority.media_alias); (0..5).all(|n| m.contains(&n)) }),
    );
    assert_eq!(media_groups(&got_clock, s_clock.media_alias), (0..5).collect::<Vec<_>>(), "the session proving the clock's key: every group");
    assert_eq!(media_groups(&got_authority, s_authority.media_alias), (0..5).collect::<Vec<_>>(), "the session proving the authority's key: every group");
    let (objects, verified, _, _) = verify(&got_clock, &s_clock, &performer);
    assert!(objects == 5 && verified >= 5, "{objects} objects, {verified} groups verified");

    // A forged proof: the clock's key named, another key's signature. The client's SETUP goes
    // out at once, so the proof names no challenge; its signature is checked first.
    let mut forged = GrantProof::sign(NO_GRANT, &stranger, [0; 32], &format!("{} {target}", grant_proof::VERB_SETUP)).expect("a proof");
    forged.key = clock.verifying_key().to_bytes();
    let forged_opts = ClientOptions { extra_setup: vec![KeyValuePair::bytes(SETUP_OPTION_GRANT_PROOF, forged.to_bytes())], ..session(vec![]) };
    let (f, _) = connect(&server, Via::WebSocket, forged_opts).await;
    closed_unauthorized(&f, "a proof naming the clock's key under another's signature").await;
    // The clock's proof made over another session's challenge.
    let replayed = ClientOptions { sign_challenge: Some([0x5e; 32]), ..session(vec![clock.clone()]) };
    let (rp, _) = connect(&server, Via::WebSocket, replayed).await;
    closed_unauthorized(&rp, "the clock's proof over another session's challenge").await;

    for c in [&p, &ordinary, &other, &by_clock, &by_authority] {
        c.close(0, "done");
    }
    server.shutdown().await;
}