//! The serving rule of WAI age-assurance §5.5 at the relay, with the ratings §5.7 carries:
//!
//! - a room rated 13 by its authority, the claim carried on the publisher's PUBLISH: a viewer
//!   whose grant states 16 (§5.4) receives every group, and one whose grant states no `age` is
//!   refused the subscription (`below_threshold`), as is a stranger's rating ignored;
//! - the publisher rates groups 5 to 7 of its own track 18, the claim carried on group 5's
//!   object, after group 4 went out: that object is not forwarded to the viewer at 16, nor,
//!   once the relay holds the claim, are groups 6 and 7, while group 8 is; a viewer at 18 is
//!   forwarded all of them;
//! - a FETCH by the viewer at 16 across the span is answered up to group 4, one that starts
//!   inside it is refused `below_threshold`; a SUBSCRIBE whose first group is inside it too;
//! - a claim a stranger signs about the room, or about the track, is not held.
//!
//! Over the WebSocket binding.

mod common;

use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::carriage::{object_properties, ObjectExtras};
use joule_moq::draft22::data::{FetchEntry, ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{Location, LocationFilter, Parameter, Parameters, RequestErrorCode};
use joule_moq::messages::ObjectHeader;
use joule_moq::rating::{self, Rating, Subject};
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::Authority;
use joule_moq_relay::RelayServer;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time::timeout;

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

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, 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 = joule_moq::grant_proof::GRANT_DOMAIN.to_vec();
    m.extend_from_slice(&g);
    g.extend_from_slice(&authority.sign(&m).to_bytes());
    g
}

fn with(server: &RelayServer, record: &[u8], key: &SigningKey) -> ClientOptions {
    ClientOptions { grants: vec![(record.to_vec(), key.clone())], target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path), ..ClientOptions::default() }
}

fn rated(age: u64) -> Vec<Rating> {
    vec![Rating { scheme: "https://rating-a.example/scheme".into(), value: age.to_string(), min_age: age }]
}

/// Group `g` of `t`, its object carrying the rating claims `claims` as Object Properties.
async fn publish_rated(c: &Client, key: &SigningKey, t: &Tracks, g: u64, payload: &[u8], claims: &[String]) {
    let (receipt, bundle) = signed_group(key, t.receipt_alias, g, payload);
    let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
    let header = ObjectHeader {
        track_alias: TrackAlias(t.media_alias),
        group_id: GroupId(g),
        object_id: ObjectId(0),
        publisher_priority: t.priority,
        status: ObjectStatus::Normal,
        payload_len: payload.len() as u64,
        mrl_layer: 0,
        mrl_total_layers: 0,
        wai_envelope: false,
    };
    let mut props = object_properties(&header, &ObjectExtras { receipt: Some(receipt), wai_receipt: None, delivery_timeout_ms: None }).unwrap();
    for c in claims {
        props.push(rating::property(c));
    }
    let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
    c.send_subgroup(mh, &[SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: payload.to_vec() }]).await.expect("the object");
}

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
}

fn range(start: u64, end: u64) -> Parameters {
    Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStartGroupEnd { start: Location::new(start, 0), end_group_delta: end - start }))
}

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

async fn run() {
    let authority = SigningKey::from_bytes(&[0x71; 32]);
    let stranger = SigningKey::from_bytes(&[0x72; 32]);
    let (performer, v0, v16, v18) = (SigningKey::from_bytes(&[0x73; 32]), SigningKey::from_bytes(&[0x74; 32]), SigningKey::from_bytes(&[0x75; 32]), SigningKey::from_bytes(&[0x76; 32]));
    let room = [0x3c; 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 t = Tracks::new(ns(&[&r, &hex::encode(performer.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
    let digest = rating::track_digest(&t.ns, &t.media);

    // The performer publishes; its PUBLISH carries the room's rating, 13, by the authority, and
    // a stranger's rating of the room at 21, which the relay does not hold.
    let (p, _) = connect(&server, Via::WebSocket, with(&server, &grant(&authority, room, &performer, 0b011, &[]), &performer)).await;
    let mut props = track_props(t.priority, None);
    props.push(rating::property(&rating::write(&authority, &Subject::Room(room), &rated(13), AT)));
    props.push(rating::property(&rating::write(&stranger, &Subject::Room(room), &rated(21), 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 claim is held, the stranger's is not");

    // A viewer whose grant states no age: refused the subscription, below the room's 13.
    let (c0, _) = connect(&server, Via::WebSocket, with(&server, &grant(&authority, room, &v0, 0, &[]), &v0)).await;
    let refused = c0.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.expect_err("below the room's rating");
    assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()) && refused.contains("below_threshold"), "{refused}");
    assert!(server.relay().stats().age_refused >= 1);

    // Viewers at 16 and 18 subscribe; groups 0 to 4 reach both.
    let (c16, _) = connect(&server, Via::WebSocket, with(&server, &grant(&authority, room, &v16, 0, &[("age", "16")]), &v16)).await;
    let (c18, _) = connect(&server, Via::WebSocket, with(&server, &grant(&authority, room, &v18, 0, &[("order", "1"), ("age", "18")]), &v18)).await;
    let s16 = subscribe_both(&c16, &t, Parameters::new()).await;
    let s18 = subscribe_both(&c18, &t, Parameters::new()).await;
    for g in 0..5 {
        publish_rated(&p, &performer, &t, g, &payload("p", g, 200), &[]).await;
    }
    // Groups 5 to 7 rated 18 by the performer, the claim on group 5's object; a stranger's
    // claim about the track at 21 rides with it, and is not honoured.
    let span = rating::write(&performer, &Subject::Span { track: digest, from_group: 5, to_group: 7 }, &rated(18), AT);
    let theirs = rating::write(&stranger, &Subject::Track { track: digest, from_group: 0 }, &rated(21), AT);
    publish_rated(&p, &performer, &t, 5, &payload("p", 5, 200), &[span, theirs]).await;
    // Each group rides a stream of its own, which the relay reads in a task of its own, so a
    // later group can reach it before the object that carries the claim: the claim holds from
    // the instant the relay reads it (§3.5), and the groups after are sent once it does. The
    // object that carries it is never forwarded below it.
    assert!(until(STEP, || server.relay().stats().ratings_held == 2).await, "the span is held");
    for g in 6..10 {
        publish_rated(&p, &performer, &t, g, &payload("p", g, 200), &[]).await;
    }
    let (got16, got18) = tokio::join!(
        collect_concurrent(&c16, STEP, Duration::from_millis(700), |g| media_groups(g, s16.media_alias).contains(&9)),
        collect_concurrent(&c18, STEP, Duration::from_millis(700), |g| media_groups(g, s18.media_alias).contains(&9)),
    );
    assert_eq!(media_groups(&got16, s16.media_alias), vec![0, 1, 2, 3, 4, 8, 9], "the viewer at 16: none of the span");
    assert_eq!(media_groups(&got18, s18.media_alias), (0..10).collect::<Vec<_>>(), "the viewer at 18: everything");
    let (objects, verified, _, _) = verify(&got16, &s16, &performer);
    assert!(objects == 7 && verified >= 7, "{objects} objects, {verified} groups verified");
    assert!(server.relay().stats().age_withheld >= 3, "{:?}", server.relay().stats());
    assert_eq!(server.relay().stats().ratings_held, 2, "the span is held; the stranger's claim about the track is not");

    // FETCH: across the span, answered up to group 4; from inside it, refused.
    let (ok, entries) = c16.fetch(t.ns.clone(), t.media.clone(), range(2, 9)).await.expect("a FETCH from group 2");
    let fetched: Vec<u64> = entries.iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some(o.group_id) } else { None }).collect();
    assert_eq!(fetched, vec![2, 3, 4], "served up to the span");
    assert_eq!(ok.end_location.group, 4, "{ok:?}");
    let refused = c16.fetch(t.ns.clone(), t.media.clone(), range(6, 9)).await.expect_err("a FETCH from inside the span");
    assert!(refused.contains("below_threshold"), "{refused}");
    let (_, all) = c18.fetch(t.ns.clone(), t.media.clone(), range(0, 9)).await.expect("the viewer at 18 fetches it all");
    assert_eq!(all.iter().filter(|e| matches!(e, FetchEntry::Object(_))).count(), 10);
    // A SUBSCRIBE whose first group is inside the span: the performer opens group 10 at 18 too.
    let more = rating::write(&performer, &Subject::Track { track: digest, from_group: 10 }, &rated(18), AT);
    publish_rated(&p, &performer, &t, 10, &payload("p", 10, 200), &[more]).await;
    assert!(until(STEP, || server.relay().stats().ratings_held == 3).await);
    let v16b = SigningKey::from_bytes(&[0x77; 32]);
    let (c16b, _) = connect(&server, Via::WebSocket, with(&server, &grant(&authority, room, &v16b, 0, &[("age", "16")]), &v16b)).await;
    let refused = c16b.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.expect_err("a live edge rated 18");
    assert!(refused.contains("below_threshold"), "{refused}");
    for c in [&p, &c0, &c16, &c16b, &c18] {
        c.close(0, "done");
    }
}