//! A room's sealed objects (WAI `extensions/confidentiality.md`) through this relay, which
//! has no notion of them: three members each publish sealed video with its receipts over
//! the sealed bytes, the door holds each publisher's namespace to its own key and checks
//! every object against its receipt, and every member opens what the others sent while a
//! subscriber outside the group verifies every receipt and opens nothing. The relay
//! forwards each object's payload and properties byte for byte.

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::draft22::data::{ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::Parameters;
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder};
use joule_moq::secure_objects::room::{Incoming, Keyring, NotOpened, Outgoing, Roster};
use joule_moq::secure_objects::{Plaintext, Refusal, Suite};
use joule_moq::{GroupId, ObjectId, TrackAlias, TrackName};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::DoorRule;

const GROUPS: u64 = 6;
const EPOCH: u64 = 3;
const ROOM: [u8; 32] = [0x5e; 32];

/// Publish group `g` of `t`: the receipts over the sealed payload first, then the sealed
/// object, whose Immutable Properties carry its Key ID and no receipt (confidentiality §4.3).
async fn publish_sealed(c: &Client, key: &SigningKey, t: &Tracks, g: u64, properties: joule_moq::draft22::kvp::Properties, payload: &[u8]) {
    let hash = ObjectReceipt::hash_payload(payload);
    let mut b = ReceiptBuilder::new(TrackAlias(t.receipt_alias), GroupId(g));
    b.add_labelled_object(hash, 0, "sealed-publisher", "2026-10-19T00:00:00Z", vec![0]).unwrap();
    let receipt = ObjectReceipt {
        track_alias: TrackAlias(t.receipt_alias),
        group_id: GroupId(g),
        object_id: ObjectId(0),
        content_hash: hash,
        joules_micro: 0,
        origin: "sealed-publisher".into(),
        merkle_proof: b.proof_for(0).unwrap(),
        energy_class: Some(vec![0]),
    };
    let group = b.finalize_labelled(key, "sealed-publisher", "2026-10-19T00:00:01Z", None).unwrap();
    let bundle = joule_moq::carriage::ReceiptBundle { group, objects: vec![receipt] };
    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 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, status: WireStatus::Normal, payload: payload.to_vec() }]).await.expect("the object");
}

async fn room_of_three(via: Via) {
    let keys: Vec<SigningKey> = [0xa1u8, 0xb2, 0xc3].iter().map(|b| SigningKey::from_bytes(&[*b; 32])).collect();
    let principals: Vec<[u8; 32]> = keys.iter().map(|k| k.verifying_key().to_bytes()).collect();
    let roster: Roster = principals.iter().enumerate().map(|(i, p)| (i as u32, *p)).collect();
    let secret = [0x77; 32];
    let rings: Vec<Keyring> = principals
        .iter()
        .map(|p| {
            let mut k = Keyring::new(Suite::AesGcm128Sha256, *p);
            k.enter(EPOCH, secret, roster.clone());
            k
        })
        .collect();
    let room = hex::encode(ROOM);
    let server = relay(|c| {
        c.receipts.door = vec![DoorRule { prefix: vec![room.clone(), "*".into(), "*".into()], trust: vec![], trust_field: Some(1), hold_ms: 5_000 }];
    })
    .await;
    let tracks: Vec<Tracks> = principals.iter().enumerate().map(|(i, p)| Tracks::new(ns(&[&room, &hex::encode(p), "video"]), "0", 10 * (i as u64 + 1), 7 + i as u64, 64)).collect();

    let mut pubs = Vec::new();
    let mut held = Vec::new();
    for t in &tracks {
        let (c, e) = connect(&server, via, ClientOptions::default()).await;
        held.push(publish_tracks(&c, t, None).await);
        pubs.push((c, e));
    }
    // The three members, and one subscriber outside the group.
    let mut subs = Vec::new();
    let mut views = Vec::new();
    for _ in 0..4 {
        let (s, e) = connect(&server, via, ClientOptions::default()).await;
        let mut v = Vec::new();
        for t in &tracks {
            v.push(subscribe_both(&s, t, Parameters::new()).await);
        }
        views.push(v);
        subs.push((s, e));
    }

    // Every member seals and publishes its groups.
    let mut sent = Vec::new();
    for (i, t) in tracks.iter().enumerate() {
        let name = TrackName::from_str("0");
        for g in 0..GROUPS {
            let pt = Plaintext::with_digest(payload(&format!("member {i}"), g, 700));
            let s = rings[i].seal(&Outgoing { ns: &t.ns, name: &name, group: g, object: 0, priority: t.priority, plaintext: &pt, immutable: &[], mutable: &[] }).unwrap().unwrap();
            publish_sealed(&pubs[i].0, &keys[i], t, g, s.properties.clone(), &s.payload).await;
            sent.push((i, g, s));
        }
    }
    let reads = futures::future::join_all(subs.iter().map(|(s, _)| collect(s, STEP * 2, |got| got.len() as u64 >= 2 * 3 * GROUPS)));
    let got = reads.await;

    for (j, (g, v)) in got.iter().zip(&views).enumerate() {
        let mut ring = if j < 3 { rings[j].clone() } else { Keyring::new(Suite::AesGcm128Sha256, [0x0f; 32]) };
        for (i, t) in tracks.iter().enumerate() {
            // Every receipt verifies over the sealed bytes the relay forwarded.
            let (objects, bundles, _, _) = verify(g, &v[i], &keys[i]);
            assert_eq!((objects, bundles), (GROUPS as usize, GROUPS as usize), "subscriber {j}, publisher {i}");
            let name = TrackName::from_str("0");
            for x in g.iter().filter(|x| x.header.track_alias == v[i].media_alias) {
                let o = &x.objects[0];
                let (_, _, s) = sent.iter().find(|(k, gg, _)| *k == i && *gg == x.header.group_id).unwrap();
                assert_eq!(o.payload, s.payload, "the sealed payload, byte for byte");
                assert_eq!(o.properties, s.properties, "the properties, the Key ID among them, byte for byte");
                let inc = Incoming { ns: &t.ns, name: &name, group: x.header.group_id, object: 0, priority: x.header.publisher_priority.unwrap(), properties: &o.properties, payload: &o.payload };
                match ring.open(&inc, None) {
                    Ok(op) if j < 3 => assert_eq!(op.plaintext.payload, payload(&format!("member {i}"), x.header.group_id, 700)),
                    Err(NotOpened::Refused(Refusal::EpochNotJoined)) if j == 3 => {}
                    other => panic!("subscriber {j}, publisher {i}: {other:?}"),
                }
            }
        }
    }
    assert_eq!(server.relay().stats().refused_at_door, 0, "every sealed object passed the door on its receipt");
    drop(held);
    for (s, _) in &subs {
        s.close(0, "done");
    }
    for (p, _) in &pubs {
        p.close(0, "done");
    }
    server.shutdown().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sealed_objects_over_websocket() {
    tokio::time::timeout(std::time::Duration::from_secs(120), room_of_three(Via::WebSocket)).await.expect("the room finishes in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sealed_objects_over_quic() {
    tokio::time::timeout(std::time::Duration::from_secs(120), room_of_three(Via::Quic)).await.expect("the room finishes in time");
}