//! A relay that serves a room whose authority issues grants as visitors arrive (WAI
//! live-rooms §10.2, §10.5): it holds the authority's key, not a rule for each grant.
//!
//! - A grant the authority signed, presented in SETUP with its proof, gives the rights
//!   §10.2 maps it onto: a publisher its own media namespaces, a viewer the room to
//!   subscribe to; the same grant with no proof closes the session, and a grant signed by
//!   another key gives nothing.
//! - The door, with a rule over `["*", "*", "video"]` and `trust_field = 1`, holds each
//!   publisher's video to the key its namespace names: a group signed by another key is
//!   dropped there.
//! - The relay's limits end a session: at its session limit (`NO_ERROR`), when its grant
//!   expires (`EXPIRED_AUTH_TOKEN`), and past its ingress limit (`INTERNAL_ERROR`).
//! - At -23 a grant that expires ends what it gave and not the session: each subscription
//!   held by it ends with PUBLISH_DONE `EXPIRED_AUTH_TOKEN`, the session stays open, and a new
//!   request by the expired grant is refused; a -22 session beside it, under a grant of the
//!   same expiry, is closed `EXPIRED_AUTH_TOKEN` as before, and so is a -23 session that
//!   publishes by the expired grant.
//!
//! Over the WebSocket binding, the transport a deployment behind an HTTP proxy uses.

mod common;

use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::draft22::{Parameters, RequestErrorCode, SessionError, Token};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DoorRule};
use joule_moq_relay::transport::TransportError;
use joule_moq_relay::RelayServer;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time::timeout;

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

/// A grant laid out as live-rooms §4.1 lays it out: `publish` the media bits, a policy that
/// permits the chat kind when `events`, signed by `authority`.
fn grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, nbf: u64, exp: u64, publish: u8, events: bool) -> Vec<u8> {
    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(&nbf.to_le_bytes());
    g.extend_from_slice(&exp.to_le_bytes());
    g.extend_from_slice(&[0, publish, 1, 0, 0, 0]);
    g.extend_from_slice(&[0; 32]);
    g.extend_from_slice(b"WEP1\x01");
    let mut kinds = [0u8; 32];
    if events {
        kinds[0] = 1 << 3;
    }
    g.extend_from_slice(&kinds);
    g.extend_from_slice(&[0; 28]);
    g.extend_from_slice(&[0xff; 16]);
    g.extend_from_slice(&[0; 3]);
    g.extend_from_slice(&[0, 0, 0, 0]);
    let mut m = joule_moq::grant_proof::GRANT_DOMAIN.to_vec();
    m.extend_from_slice(&g);
    g.extend_from_slice(&authority.sign(&m).to_bytes());
    g
}

fn target(server: &RelayServer) -> String {
    format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path)
}

fn with(server: &RelayServer, record: &[u8], key: &SigningKey) -> ClientOptions {
    ClientOptions { grants: vec![(record.to_vec(), key.clone())], target: target(server), ..ClientOptions::default() }
}

async fn closed_with(c: &Client, code: SessionError, within: Duration, what: &str) -> String {
    match timeout(within, c.conn.closed()).await.unwrap_or_else(|_| panic!("{what}: not closed in time")) {
        TransportError::Closed { code: got, reason, .. } => {
            assert_eq!(got, code.0, "{what}: {reason}");
            reason
        }
        other => panic!("{what}: {other:?}"),
    }
}

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

async fn run() {
    let authority = SigningKey::from_bytes(&[0x61; 32]);
    let stranger = SigningKey::from_bytes(&[0x62; 32]);
    let (alice, bob, viewer) = (SigningKey::from_bytes(&[0x63; 32]), SigningKey::from_bytes(&[0x64; 32]), SigningKey::from_bytes(&[0x65; 32]));
    let room = [0x7e; 32];
    let r = hex::encode(room);
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: hex::encode(authority.verifying_key().to_bytes()), rooms: vec![] }];
        c.receipts.door = vec![DoorRule { prefix: vec!["*".into(), "*".into(), "video".into()], trust: vec![], trust_field: Some(1), hold_ms: 2_000 }];
    })
    .await;
    let now = now_s();
    let ga = grant(&authority, room, &alice, now - 60, now + 600, 0b011, true);
    let gb = grant(&authority, room, &bob, now - 60, now + 600, 0b011, true);
    let gv = grant(&authority, room, &viewer, now - 60, now + 600, 0, false);
    let ta = Tracks::new(ns(&[&r, &hex::encode(alice.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
    let tb = Tracks::new(ns(&[&r, &hex::encode(bob.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);

    // Alice publishes her video; the viewer subscribes to it and every object verifies.
    let (a, _) = connect(&server, Via::WebSocket, with(&server, &ga, &alice)).await;
    let _pa = publish_tracks(&a, &ta, None).await;
    let (v, _) = connect(&server, Via::WebSocket, with(&server, &gv, &viewer)).await;
    let sub = subscribe_both(&v, &ta, Parameters::new()).await;
    for g in 0..5 {
        publish_group(&a, &alice, &ta, g, &payload("a", g, 300)).await;
    }
    let got = collect_concurrent(&v, STEP, Duration::from_millis(500), |g| g.iter().filter(|x| x.header.track_alias == sub.media_alias).count() >= 5).await;
    let (objects, verified, _, _) = verify(&got, &sub, &alice);
    assert!(objects >= 5 && verified == objects, "{verified} of {objects} verified");

    // Alice may not publish under Bob's namespace, the viewer may publish nothing of media,
    // and neither may publish in another room.
    let refused = a.publish(tb.ns.clone(), tb.media.clone(), 30, Parameters::new(), track_props(128, None)).await.expect_err("another principal's namespace");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
    let refused = v.publish(ns(&[&r, &hex::encode(viewer.verifying_key().to_bytes()), "video"]), name("0"), 30, Parameters::new(), track_props(128, None)).await.expect_err("a viewer's media");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");

    // The door: Bob publishes in his own namespace with Alice's key; his group is dropped.
    let (b, _) = connect(&server, Via::WebSocket, with(&server, &gb, &bob)).await;
    let _pb = publish_tracks(&b, &tb, None).await;
    let sb = subscribe_both(&v, &tb, Parameters::new()).await;
    publish_group(&b, &alice, &tb, 0, &payload("b", 0, 300)).await;
    let got = collect_concurrent(&v, Duration::from_secs(4), Duration::from_millis(500), |_| false).await;
    assert!(got.iter().all(|x| x.header.track_alias != sb.media_alias || x.reset.is_some() || x.objects.is_empty()), "a group signed by another key went past the door");
    assert!(server.relay().stats().refused_at_door >= 1 || server.relay().stats().groups_dropped_at_door >= 1, "the door refused it");

    // The grant with no proof: the session is refused. A grant another key signed is no
    // grant here: its requests are refused, its session stands.
    let bare = ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: ga.clone() }], ..ClientOptions::default() };
    let (c, _) = connect(&server, Via::WebSocket, bare).await;
    closed_with(&c, SessionError::UNAUTHORIZED, STEP, "a grant with no proof").await;
    let forged = grant(&stranger, room, &alice, now - 60, now + 600, 0b011, true);
    let (c, _) = connect(&server, Via::WebSocket, with(&server, &forged, &alice)).await;
    let refused = c.subscribe(ta.ns.clone(), ta.media.clone(), Parameters::new()).await.expect_err("a grant another key signed");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
    for c in [&a, &b, &v, &c] {
        c.close(0, "done");
    }
    server.shutdown().await;

    // The limits, each on a relay of its own.
    let limited = |f: fn(&mut joule_moq_relay::RelayConfig)| {
        let key = hex::encode(authority.verifying_key().to_bytes());
        relay(move |c| {
            c.auth.required = true;
            c.auth.authorities = vec![Authority { key, rooms: vec![] }];
            f(c);
        })
    };
    // At its session limit; a session whose token's rule is `unlimited` is not.
    let s = limited(|c| {
        c.limits.max_session_ms = 1_500;
        c.auth.grants = vec![joule_moq_relay::config::Grant { prefix: vec![], token_sha256: hex::encode(joule_moq_relay::auth::sha256(b"service")), publish: false, subscribe: true, proof: false, unlimited: true, seek: false }];
    })
    .await;
    let (c, _) = connect(&s, Via::WebSocket, with(&s, &gv, &viewer)).await;
    let (svc, _) = connect(&s, Via::WebSocket, ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: b"service".to_vec() }], ..ClientOptions::default() }).await;
    let why = closed_with(&c, SessionError::NO_ERROR, Duration::from_secs(10), "the session limit").await;
    assert!(why.contains("session limit"), "{why}");
    assert!(timeout(Duration::from_secs(2), svc.conn.closed()).await.is_err(), "an unlimited session outlives the limit");
    svc.close(0, "done");
    s.shutdown().await;
    // When its grant expires.
    let s = limited(|_| {}).await;
    let short = grant(&authority, room, &viewer, now - 60, now_s() + 2, 0, false);
    let (c, _) = connect(&s, Via::WebSocket, with(&s, &short, &viewer)).await;
    closed_with(&c, SessionError::EXPIRED_AUTH_TOKEN, Duration::from_secs(10), "the grant's expiry").await;
    s.shutdown().await;
    // Past its ingress limit: 64 kbit/s over a second is 8 kB; ten 4 kB groups pass it.
    let s = limited(|c| {
        c.limits.max_ingress_kbps = 64;
        c.limits.ingress_burst_ms = 1_000;
    })
    .await;
    let (c, _) = connect(&s, Via::WebSocket, with(&s, &ga, &alice)).await;
    let _p = publish_tracks(&c, &ta, None).await;
    for g in 0..10 {
        if c.send_subgroup(
            joule_moq::draft22::data::SubgroupHeader { track_alias: ta.media_alias, group_id: g, subgroup_id: joule_moq::draft22::data::SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true },
            &[joule_moq::draft22::data::SubgroupObject::normal(0, payload("x", g, 4_000))],
        )
        .await
        .is_err()
        {
            break;
        }
    }
    // The relay closes the session while the client may still be writing; then the client
    // sees its pipe break before the close arrives, and the relay's own record says why.
    match timeout(Duration::from_secs(10), c.conn.closed()).await.expect("closed in time") {
        TransportError::Closed { code, reason, .. } => {
            assert_eq!(code, SessionError::INTERNAL_ERROR.0, "{reason}");
            assert!(reason.contains("ingress limit"), "{reason}");
        }
        other => {
            let closed = until(Duration::from_secs(10), || s.relay().summaries().iter().any(|x| x.summary.close_code == Some(SessionError::INTERNAL_ERROR.0))).await;
            assert!(closed, "the relay closed the session past its ingress limit (the client saw {other:?})");
        }
    }
    s.shutdown().await;
}

/// A grant's expiry at -23 and at -22, side by side: three sessions under grants that expire
/// together. Planted: the older drafts' behaviour at -23 (the session closed, no PUBLISH_DONE)
/// fails the first viewer's checks, and an expiry that ended nothing fails its refused request.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_grant_that_expires_ends_its_subscription_at_23_and_the_session_at_22() {
    timeout(Duration::from_secs(120), expiry()).await.expect("the test finishes in time");
}

async fn expiry() {
    use joule_moq::draft::Draft;
    use joule_moq::draft22::ControlMessage;
    let authority = SigningKey::from_bytes(&[0x71; 32]);
    let (alice, bob, v23, v22) = (SigningKey::from_bytes(&[0x73; 32]), SigningKey::from_bytes(&[0x74; 32]), SigningKey::from_bytes(&[0x75; 32]), SigningKey::from_bytes(&[0x76; 32]));
    let room = [0x7d; 32];
    let r = hex::encode(room);
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: hex::encode(authority.verifying_key().to_bytes()), rooms: vec![] }];
    })
    .await;
    let now = now_s();
    let soon = now + 4;
    let ga = grant(&authority, room, &alice, now - 60, now + 600, 0b011, false);
    let ta = Tracks::new(ns(&[&r, &hex::encode(alice.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
    let tb = Tracks::new(ns(&[&r, &hex::encode(bob.verifying_key().to_bytes()), "video"]), "0", 20, 6, 128);
    let (a, _) = connect(&server, Via::WebSocket, with(&server, &ga, &alice)).await;
    let _pa = publish_tracks(&a, &ta, None).await;
    // Three sessions under grants that expire at one second: a viewer at -23, a viewer at -22,
    // and a publisher at -23.
    let (c23, _) = connect(&server, Via::WsDraft(Draft::D23), with(&server, &grant(&authority, room, &v23, now - 60, soon, 0, false), &v23)).await;
    let (c22, _) = connect(&server, Via::WebSocket, with(&server, &grant(&authority, room, &v22, now - 60, soon, 0, false), &v22)).await;
    let (b, _) = connect(&server, Via::WsDraft(Draft::D23), with(&server, &grant(&authority, room, &bob, now - 60, soon, 0b011, false), &bob)).await;
    let _pb = publish_tracks(&b, &tb, None).await;
    let mut s23 = subscribe_both(&c23, &ta, Parameters::new()).await;
    let _s22 = subscribe_both(&c22, &ta, Parameters::new()).await;
    publish_group(&a, &alice, &ta, 0, &payload("a", 0, 300)).await;
    let got = collect(&c23, STEP, |x| x.len() >= 2).await;
    assert_eq!(got.len(), 2, "the -23 viewer is served while its grant is good");
    assert_eq!(server.relay().stats().sessions_open, 4);

    // The grants expire. At -23 each subscription the grant gave ends with PUBLISH_DONE
    // EXPIRED_AUTH_TOKEN (0x7), the media track's and the receipts track's.
    for (_, rx) in s23.held.iter_mut() {
        match timeout(Duration::from_secs(15), rx.next()).await.expect("a PUBLISH_DONE at the grant's expiry") {
            Ok(Some(ControlMessage::PublishDone(d))) => assert_eq!(d.status.0, 0x7, "EXPIRED_AUTH_TOKEN: {:?}", d.status),
            other => panic!("the subscription ended otherwise: {other:?}"),
        }
    }
    // At -22 the session ends, as before; and so does the -23 session that publishes by its
    // grant, which no PUBLISH_DONE of the relay's can end.
    closed_with(&c22, SessionError::EXPIRED_AUTH_TOKEN, Duration::from_secs(10), "the -22 session at its grant's expiry").await;
    closed_with(&b, SessionError::EXPIRED_AUTH_TOKEN, Duration::from_secs(10), "the -23 session that publishes by the expired grant").await;
    // The -23 viewer's session goes on: open, sent no GOAWAY, forwarded nothing more, and its
    // expired grant gives it no new subscription.
    assert!(timeout(Duration::from_millis(1500), c23.conn.closed()).await.is_err(), "the -23 session was closed at its grant's expiry");
    assert!(c23.goaway().borrow().is_none());
    publish_group(&a, &alice, &ta, 1, &payload("a", 1, 300)).await;
    let more = collect(&c23, Duration::from_secs(2), |x| !x.is_empty()).await;
    assert!(more.is_empty(), "an object forwarded on a subscription the expired grant gave: {more:?}");
    let refused = c23.subscribe(ta.ns.clone(), ta.media.clone(), Parameters::new()).await.expect_err("a subscription by an expired grant");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
    assert!(until(STEP, || server.relay().stats().sessions_open == 2).await, "the publisher and the -23 viewer, open");
    a.close(0, "done");
    c23.close(0, "done");
    server.shutdown().await;
}