//! A rating that rises while a subscriber's objects wait at the relay (WAI age-assurance §3.5,
//! §5.5): "from the instant it holds a claim that raises a group's age above the session's
//! threshold, the relay forwards none of that group's objects to the session".
//!
//! A viewer at 18 subscribes over native QUIC with small receive windows and reads nothing,
//! and the relay is given a small send window, so the performer's groups 0 to 9, each 24 KiB,
//! wait in the relay's scheduler for that viewer. Then the performer rates its own track 21
//! from group 0 on, the claim on group 10's object. Once the relay holds it, the viewer reads
//! everything: of groups 0 to 9 it is given only what the transport already held before the
//! claim, those still waiting at the relay ended (`CANCELLED`) and sent no more, and of group 10
//! nothing. A viewer at 21 is given every group.

mod common;

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

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

fn grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, age: Option<&str>) -> Vec<u8> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
    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> = age.map(|v| [&[3u8][..], b"age", &(v.len() as u16).to_le_bytes(), v.as_bytes()].concat()).unwrap_or_default();
    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 rated(age: u64) -> Vec<Rating> {
    vec![Rating { scheme: "https://rating-a.example/scheme".into(), value: age.to_string(), min_age: age }]
}

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

async fn run() {
    let authority = SigningKey::from_bytes(&[0x81; 32]);
    let (performer, slow_key, fast_key) = (SigningKey::from_bytes(&[0x82; 32]), SigningKey::from_bytes(&[0x83; 32]), SigningKey::from_bytes(&[0x84; 32]));
    let room = [0x4d; 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.limits.quic_send_window_bytes = 64 * 1024;
    })
    .await;
    let quic = |record: Vec<u8>, key: &SigningKey, lazy: bool| ClientOptions { grants: vec![(record, key.clone())], target: "//".into(), lazy_data: lazy, ..ClientOptions::default() };
    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);

    let (p, _ep) = connect(&server, Via::Quic, quic(grant(&authority, room, &performer, 0b011, None), &performer, false)).await;
    let mut props = track_props(t.priority, None);
    props.push(rating::property(&rating::write(&authority, &Subject::Room(room), &rated(13), 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");

    let mut small = quinn::TransportConfig::default();
    small.stream_receive_window(quinn::VarInt::from_u32(32 * 1024));
    small.receive_window(quinn::VarInt::from_u32(96 * 1024));
    let (slow, _es) = connect_with(&server, Via::Quic, quic(grant(&authority, room, &slow_key, 0, Some("18")), &slow_key, true), Some(small)).await;
    let (fast, _ef) = connect(&server, Via::Quic, quic(grant(&authority, room, &fast_key, 0, Some("21")), &fast_key, false)).await;
    let next = || Parameters::new().with(Parameter::LocationFilter(LocationFilter::NextObject));
    let s_slow = subscribe_both(&slow, &t, next()).await;
    let s_fast = subscribe_both(&fast, &t, next()).await;

    // Until every group has arrived (each on a stream of its own, in any order).
    let fast_reads = collect_concurrent(&fast, STEP * 2, Duration::from_secs(1), |g| (0..=GROUPS).all(|n| g.iter().any(|x| x.header.track_alias == s_fast.media_alias && x.header.group_id == n && x.reset.is_none() && !x.objects.is_empty())));
    let publish = async {
        for g in 0..GROUPS {
            publish_group(&p, &performer, &t, g, &payload("p", g, 24 * 1024)).await;
        }
        // Group 10's object carries the performer's rating of its whole track at 21.
        let (receipt, bundle) = signed_group(&performer, t.receipt_alias, GROUPS, &payload("p", GROUPS, 1024));
        let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: GROUPS, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
        p.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(GROUPS),
            object_id: ObjectId(0),
            publisher_priority: t.priority,
            status: ObjectStatus::Normal,
            payload_len: 1024,
            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();
        props.push(rating::property(&rating::write(&performer, &Subject::Track { track: digest, from_group: 0 }, &rated(21), AT)));
        let mh = SubgroupHeader { track_alias: t.media_alias, group_id: GROUPS, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
        p.send_subgroup(mh, &[SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: payload("p", GROUPS, 1024) }]).await.expect("the object");
        assert!(until(STEP, || server.relay().stats().ratings_held == 2).await, "the rise is held");
    };
    let ((), fast_got) = tokio::join!(publish, fast_reads);
    // Only now does the slow viewer read.
    // It reads until every receipt has come (the receipts track is not the one the rise rates,
    // so all of it is forwarded) and a second passes with nothing new.
    let receipts = s_slow.receipts_alias;
    let slow_got = collect_concurrent(&slow, STEP * 2, Duration::from_secs(1), |g| (0..=GROUPS).all(|n| g.iter().any(|x| x.header.track_alias == receipts && x.header.group_id == n && x.reset.is_none()))).await;

    let whole = |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
    };
    let fast_groups = whole(&fast_got, s_fast.media_alias);
    assert_eq!(fast_groups, (0..=GROUPS).collect::<Vec<_>>(), "the viewer at 21: every group");
    let slow_groups = whole(&slow_got, s_slow.media_alias);
    assert!(!slow_groups.contains(&GROUPS), "the object that carries the rise is not forwarded below it: {slow_groups:?}");
    assert!(slow_groups.len() < GROUPS as usize, "groups that waited at the relay when the rise was held were sent no more: {slow_groups:?}");
    let ended = slow_got.iter().filter(|x| x.header.track_alias == s_slow.media_alias && x.reset == Some(joule_moq::draft22::StreamResetCode::CANCELLED.0)).count();
    let delivered_or_ended = slow_groups.len() + ended;
    assert!(ended > 0 || delivered_or_ended < GROUPS as usize, "{ended} streams ended, {} groups whole", slow_groups.len());
    // What it was given verifies, against receipts that kept flowing.
    let (objects, _, _, _) = verify(&slow_got, &s_slow, &performer);
    assert_eq!(objects, slow_groups.len());
    for c in [&p, &slow, &fast] {
        c.close(0, "done");
    }
}