//! A room with a shared world (WAI live-worlds), kept by the room service, against a relay on
//! loopback that trusts its authority and holds every world track to its owner's key:
//!
//! - the room carries the authority's world statement, which checks against its descriptor;
//!   every publisher's grant lets it act (`world` = `act`) and publishes no media;
//! - two actors publish their world tracks, each group an op batch signed into the actor's
//!   receipt chain as the room client's signer signs it; the clock signs each actor's `join`
//!   when its world track appears, and no `publish`, and its `leave` when the session ends;
//! - another actor that enters later fetches both tracks from group 0, and its copy of the
//!   world, every receipt and op checked by `wai-rs`'s live-worlds module, holds the session
//!   hash of every op the two published, which it shares with each of them;
//! - a viewer's grant publishes no world track, an actor's publishes no other principal's, and
//!   the relay's door drops a group whose receipt is not its owner's.

use ed25519_dalek::SigningKey;
use joule_moq::carriage::{object_properties, ObjectExtras, ReceiptBundle, WaiReceiptProperty};
use joule_moq::draft22::data::{FetchEntry, ObjectStatus as WireStatus, SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
use joule_moq::draft22::{Parameters, Token};
use joule_moq::messages::ObjectHeader;
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder};
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias, TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DoorRule, Grant};
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::websocket;
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, Config, Service};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;
use wai::live_world::{self, World, RECEIPT_ALIAS};

const STEP: Duration = Duration::from_secs(20);
const MODEL: &str = "0707070707070707070707070707070707070707070707070707070707070707";
const WORLD_PRIORITY: u8 = 48;

async fn relay(keys: &Keys) -> RelayServer {
    let mut c = RelayConfig::loopback();
    c.quic_listen = None;
    c.auth.required = true;
    c.auth.authorities = vec![Authority { key: hex::encode(keys.authority.verifying_key().to_bytes()), rooms: vec![] }];
    let t = hex::encode(Sha256::digest(keys.clock_token));
    let clock = hex::encode(keys.clock.verifying_key().to_bytes());
    c.auth.grants = vec![
        Grant { prefix: vec!["*".into(), clock, "events".into()], token_sha256: t.clone(), publish: true, subscribe: false, proof: false, unlimited: true, seek: false },
        Grant { prefix: vec![], token_sha256: t, publish: false, subscribe: true, proof: false, unlimited: false, seek: false },
    ];
    // The relay-lines' world door (live-worlds §4.3).
    c.receipts.door = vec![DoorRule { prefix: vec!["*".into(), "*".into(), "world".into()], trust: vec![], trust_field: Some(1), hold_ms: 2_000 }];
    RelayServer::start(c).await.expect("the relay starts")
}

async fn call(addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>, client: &str) -> (u16, Value) {
    let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
    let b = body.map(|v| v.to_string()).unwrap_or_default();
    let req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\nCF-Connecting-IP: {client}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}", b.len());
    s.write_all(req.as_bytes()).await.unwrap();
    let mut out = Vec::new();
    timeout(STEP, s.read_to_end(&mut out)).await.expect("an answer in time").unwrap();
    let text = String::from_utf8(out).unwrap();
    let (head, body) = text.split_once("\r\n\r\n").expect("a response");
    let status: u16 = head.split(' ').nth(1).unwrap().parse().unwrap();
    (status, serde_json::from_str(body).unwrap_or(Value::Null))
}

fn key(i: u8) -> SigningKey {
    SigningKey::from_bytes(&[i; 32])
}

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

async fn ask(addr: std::net::SocketAddr, k: &SigningKey, role: &str, client: &str) -> (u16, Value) {
    call(addr, "POST", "/grant", Some(&serde_json::json!({"principal": pubhex(k), "role": role})), client).await
}

async fn session(server: &RelayServer, opts: ClientOptions) -> Client {
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let dial = || async { Ok::<joule_moq_relay::transport::SharedConnection, String>(Arc::new(websocket::connect(&url, MuxConfig::default()).await.map_err(|e| e.to_string())?)) };
    // A session that stays to see the relay refuse it: one SETUP, as it is.
    if opts.ignore_refusal {
        return Client::connect(dial().await.expect("a WebSocket session"), opts).await.expect("SETUP");
    }
    // The relay reads this SETUP before it sends its own, so the grant is proved over a
    // challenge its SETUP gave: on a second session, where the first held none.
    Client::connect_proving(dial, opts).await.expect("SETUP")
}

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

/// An actor's world track as the room client publishes it (joule-moq-web's signer): each group
/// one op batch, signed under the actor's key into a group receipt chained to the group before,
/// the unmetered marker its figure; the receipts-track stream, then the object with its
/// `WAI_RECEIPT` property and no envelope flag.
struct WorldTrack {
    key: SigningKey,
    alias: u64,
    receipts_alias: u64,
    next: u64,
    parent: Option<[u8; 32]>,
    _held: Vec<(joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)>,
}

impl WorldTrack {
    async fn publish(c: &Client, room: &str, k: &SigningKey, alias: u64) -> Result<WorldTrack, String> {
        let ns = TrackNamespace::from_strs([room, pubhex(k).as_str(), "world"]);
        let mut held = Vec::new();
        for (name, a) in [("0_receipts", alias + 1), ("0", alias)] {
            let mut p = Properties::new();
            p.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, WORLD_PRIORITY as u64));
            held.push(c.publish(ns.clone(), TrackName::from_str(name), a, Parameters::new(), p).await?);
        }
        Ok(WorldTrack { key: k.clone(), alias, receipts_alias: alias + 1, next: 0, parent: None, _held: held })
    }

    /// The two streams of the next group, holding `batch`, its receipt signed by `signer`.
    fn sign(&mut self, batch: &[u8], signer: &SigningKey) -> (Vec<u8>, Vec<u8>) {
        let g = self.next;
        let now = "2026-10-25T00:00:00Z";
        let hash = ObjectReceipt::hash_payload(batch);
        let mut b = ReceiptBuilder::new(TrackAlias(RECEIPT_ALIAS), GroupId(g));
        b.add_labelled_object(hash, 0, "world-test", now, vec![0]).unwrap();
        let receipt = ObjectReceipt {
            track_alias: TrackAlias(RECEIPT_ALIAS),
            group_id: GroupId(g),
            object_id: ObjectId(0),
            content_hash: hash,
            joules_micro: 0,
            origin: "world-test".into(),
            merkle_proof: b.proof_for(0).unwrap(),
            energy_class: Some(vec![0]),
        };
        let gr = b.finalize_labelled(signer, "world-test", now, self.parent).unwrap();
        let wai = WaiReceiptProperty::from_receipts(&receipt, &gr).expect("a WAI_RECEIPT property");
        let link = gr.link_hash();
        let receipts = ReceiptBundle { group: gr, objects: vec![receipt] }.encode_stream(self.receipts_alias, g, WORLD_PRIORITY).unwrap();
        let header = ObjectHeader {
            track_alias: TrackAlias(self.alias),
            group_id: GroupId(g),
            object_id: ObjectId(0),
            publisher_priority: WORLD_PRIORITY,
            status: ObjectStatus::Normal,
            payload_len: batch.len() as u64,
            mrl_layer: 0,
            mrl_total_layers: 0,
            wai_envelope: false,
        };
        let props = object_properties(&header, &ObjectExtras { receipt: None, wai_receipt: Some(wai), delivery_timeout_ms: None }).unwrap();
        let sh = SubgroupHeader { track_alias: self.alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(WORLD_PRIORITY), properties: true, end_of_group: true, first_object: true };
        let mut object = Vec::new();
        sh.encode(&mut object);
        SubgroupCodec::new(sh).encode_object(&SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: batch.to_vec() }, &mut object).unwrap();
        self.parent = Some(link);
        self.next += 1;
        (receipts, object)
    }

    async fn send(&mut self, c: &Client, batch: &[u8]) {
        let k = self.key.clone();
        self.send_signed_by(c, batch, &k).await;
    }

    async fn send_signed_by(&mut self, c: &Client, batch: &[u8], signer: &SigningKey) {
        let (r, o) = self.sign(batch, signer);
        for b in [r, o] {
            let mut s = timeout(STEP, c.conn.open_uni()).await.expect("a stream in time").expect("a stream");
            s.write_all(&b).await.expect("written");
            let _ = s.finish();
        }
    }
}

fn act(k: &SigningKey, l: u64, dx: i32, dy: i32) -> Vec<u8> {
    wai::world_model::sign_action(k, l, l, dx, dy, 0, b"").to_bytes()
}

/// `owner`'s world track and its receipts track, fetched from group 0 (live-worlds §7): each
/// group's receipts-track object and op batch.
async fn fetch_track(c: &Client, room: &str, owner: &str) -> Vec<(u64, Vec<u8>, Vec<u8>)> {
    let ns = TrackNamespace::from_strs([room, owner, "world"]);
    let objects = |entries: Vec<FetchEntry>| -> std::collections::BTreeMap<u64, Vec<u8>> {
        entries.into_iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some((o.group_id, o.payload)) } else { None }).collect()
    };
    let (_, r) = c.fetch(ns.clone(), TrackName::from_str("0_receipts"), Parameters::new()).await.expect("FETCH the receipts track");
    let (_, o) = c.fetch(ns, TrackName::from_str("0"), Parameters::new()).await.expect("FETCH the world track");
    let (r, o) = (objects(r), objects(o));
    o.into_iter().filter_map(|(g, batch)| r.get(&g).map(|rc| (g, rc.clone(), batch))).collect()
}

/// The clock's lifecycle kinds as the relay serves its events track.
async fn kinds(c: &Client, room: &str, clock: &str, t0_us: u64) -> Vec<String> {
    let ns = TrackNamespace::from_strs([room, clock, "events"]);
    let Ok((_, entries)) = c.fetch(ns, TrackName::from_str("0"), Parameters::new()).await else { return vec![] };
    let mut by_seq = std::collections::BTreeMap::new();
    for e in entries {
        let FetchEntry::Object(o) = e else { continue };
        let arr: Vec<Value> = serde_json::from_slice(&o.payload).expect("a JSON array");
        for r in arr {
            let rec = wai::live_events::from_msf_record(&r.to_string(), t0_us).expect("an MSF record");
            let ev = wai::live_events::Event::parse(&rec).expect("a WEV1 record");
            by_seq.insert(ev.seq, wai::live_events::kind_name(ev.kind).unwrap().to_string());
        }
    }
    by_seq.into_values().collect()
}

async fn until_kinds(c: &Client, room: &str, clock: &str, t0: u64, want: &[&str]) -> Vec<String> {
    let mut k = Vec::new();
    for _ in 0..150 {
        k = kinds(c, room, clock, t0).await;
        if k.len() >= want.len() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    k
}

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

async fn run() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\nmax_publishers = 4\njoin_wait_s = 30\n[world]\nmodel_id = \"{MODEL}\"\nwidth = 64\nheight = 64\nfps = 24\n",
        server.websocket_addr.unwrap(),
        server.relay().config().path
    ))
    .unwrap();
    let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
    let listener = tokio::net::TcpListener::bind(cfg.listen).await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(svc.clone().run_clock());
    tokio::spawn(http::serve(svc.clone(), listener));
    timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("the clock's session opens");

    // Alice enters: her grant lets her act and publishes no media; the room carries its statement.
    let alice = key(0x31);
    let (st, a) = ask(addr, &alice, "publisher", "10.0.0.1").await;
    assert_eq!(st, 200, "{a}");
    let room = a["room"]["room"].as_str().unwrap().to_string();
    let t0 = a["room"]["t0_us"].as_u64().unwrap();
    let descriptor = hex::decode(a["room"]["descriptor"].as_str().unwrap()).unwrap();
    let statement = hex::decode(a["room"]["world"]["statement"].as_str().expect("the room's world statement")).unwrap();
    assert_eq!(a["room"]["world"]["receipt_alias"].as_u64(), Some(RECEIPT_ALIAS));
    let d = wai::live_room::Descriptor::parse(&descriptor).unwrap();
    let s = live_world::Statement::check(&statement, &d, &d.room_id()).expect("the statement checks against the descriptor");
    assert_eq!(hex::encode(s.header().model_id), MODEL);
    let ga = hex::decode(a["grant"].as_str().unwrap()).unwrap();
    let held = wai::live_room::check_grants(&d, &d.room_id(), std::slice::from_ref(&ga)).expect("the grant checks");
    assert!(live_world::acts(&held[0].grant.attrs) && held[0].grant.caps.publish == 0, "an actor's grant: world = act, no media");

    let ca = session(&server, with_grant(&server, &ga, &alice)).await;
    let mut ta = WorldTrack::publish(&ca, &room, &alice, 40).await.expect("Alice publishes her world track");
    // She publishes no media and no other principal's world track.
    let bob = key(0x32);
    assert!(WorldTrack::publish(&ca, &room, &bob, 60).await.is_err(), "another principal's world track is refused");
    let ns_video = TrackNamespace::from_strs([room.as_str(), pubhex(&alice).as_str(), "video"]);
    assert!(ca.publish(ns_video, TrackName::from_str("0"), 70, Parameters::new(), Properties::new()).await.is_err(), "an actor's grant publishes no media");
    let ops_a = [act(&alice, 1, 1, 0), act(&alice, 2, 0, 1), act(&alice, 3, 1, 1)];
    ta.send(&ca, &ops_a[0]).await;
    ta.send(&ca, &[ops_a[1].clone(), ops_a[2].clone()].concat()).await;

    // The clock records her joining when her world track appears, and signs no publish.
    let clock = pubhex(&keys.clock);
    let viewer = key(0x33);
    let (st, v) = ask(addr, &viewer, "viewer", "10.0.0.3").await;
    assert_eq!(st, 200);
    let gv = hex::decode(v["grant"].as_str().unwrap()).unwrap();
    let cv = session(&server, with_grant(&server, &gv, &viewer)).await;
    assert_eq!(until_kinds(&cv, &room, &clock, t0, &["join"]).await, ["join"]);
    // A viewer's grant publishes no world track.
    assert!(WorldTrack::publish(&cv, &room, &viewer, 40).await.is_err(), "a viewer's grant publishes no world track");

    // Bob enters and acts; his last group is receipted by Alice's key, and the relay's door
    // drops it.
    let (st, b) = ask(addr, &bob, "publisher", "10.0.0.2").await;
    assert_eq!(st, 200, "{b}");
    let gb = hex::decode(b["grant"].as_str().unwrap()).unwrap();
    let cb = session(&server, with_grant(&server, &gb, &bob)).await;
    let mut tb = WorldTrack::publish(&cb, &room, &bob, 40).await.expect("Bob publishes his world track");
    let ops_b = [act(&bob, 1, -1, 0), act(&bob, 2, 0, -1)];
    for o in &ops_b {
        tb.send(&cb, o).await;
    }
    tb.send_signed_by(&cb, &act(&bob, 3, 5, 5), &alice).await;
    assert_eq!(until_kinds(&cv, &room, &clock, t0, &["join", "join"]).await, ["join", "join"]);

    // Carol enters later and catches up: both tracks fetched from group 0, every receipt and
    // op checked by wai-rs's live-worlds module.
    let carol = key(0x34);
    let (st, c) = ask(addr, &carol, "publisher", "10.0.0.4").await;
    assert_eq!(st, 200, "{c}");
    let gc = hex::decode(c["grant"].as_str().unwrap()).unwrap();
    let cc = session(&server, with_grant(&server, &gc, &carol)).await;
    tokio::time::sleep(Duration::from_millis(2_500)).await; // past the door's hold for Bob's last group
    let grants: Vec<Vec<u8>> = c["room"]["grants"].as_array().unwrap().iter().map(|g| hex::decode(g.as_str().unwrap()).unwrap()).collect();
    let held = wai::live_room::check_grants(&d, &d.room_id(), &grants).expect("the room's grants check");
    let mut world = World::new(&d, &held, &statement, false).expect("the room's world");
    for owner in [&alice, &bob] {
        let groups = fetch_track(&cc, &room, &pubhex(owner)).await;
        assert_eq!(groups.iter().map(|g| g.0).collect::<Vec<u64>>(), vec![0, 1], "every group the door passed, from 0; none it dropped");
        for (g, receipts, batch) in groups {
            world.offer(&owner.verifying_key().to_bytes(), g, &receipts, &batch, None).unwrap_or_else(|r| panic!("group {g} of {}: {r}", pubhex(owner)));
        }
    }
    assert_eq!(world.count(), 5);
    // Each actor's own copy (its ops and the other's) holds the same session hash.
    let all: Vec<wai::world::CausalOp> = ops_a.iter().chain(ops_b.iter()).map(|b| wai::world::CausalOp::from_bytes(b).unwrap().0).collect();
    let header = s.header();
    let want = wai::world_model::replay(&header, &all, true).unwrap().session_hash;
    assert_eq!(world.session_hash(), want, "the late joiner's world is the actors' world");
    let mut reversed = all.clone();
    reversed.reverse();
    assert_eq!(wai::world_model::replay(&header, &reversed, true).unwrap().session_hash, want, "whatever order the ops arrived in");

    // A grant alone, without its proof, opens no session (live-rooms §10.5).
    let bare = session(&server, ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: gc.clone() }], ignore_refusal: true, ..ClientOptions::default() }).await;
    assert!(timeout(STEP, bare.conn.closed()).await.is_ok(), "the relay closes a session whose grant has no proof");

    // Alice leaves; the clock says so.
    drop(ta);
    ca.close(0, "done");
    assert_eq!(until_kinds(&cv, &room, &clock, t0, &["join", "join", "leave"]).await, ["join", "join", "leave"]);
    // Her ops stay part of the world: a page entering now still fetches her track from 0.
    let groups = fetch_track(&cc, &room, &pubhex(&alice)).await;
    assert_eq!(groups.len(), 2, "a departed actor's track is still fetched");
    drop((tb, cb, cc, cv));
}

/// A private world (live-worlds §8): the service keeps an end-to-end encrypted world room, its
/// confidentiality record and its world statement both served, and an actor's `join` brings the
/// clock's proposal to add it on the room's log.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_private_world_room_keys_its_actors() {
    timeout(Duration::from_secs(120), private()).await.expect("the test finishes in time");
}

async fn private() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\ne2ee = true\njoin_wait_s = 30\n[world]\nmodel_id = \"{MODEL}\"\nwidth = 64\nheight = 64\nfps = 24\n",
        server.websocket_addr.unwrap(),
        server.relay().config().path
    ))
    .unwrap();
    let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
    let listener = tokio::net::TcpListener::bind(cfg.listen).await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(svc.clone().run_clock());
    tokio::spawn(http::serve(svc.clone(), listener));
    timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("the clock's session opens");
    let alice = key(0x41);
    let (st, a) = ask(addr, &alice, "publisher", "10.0.1.1").await;
    assert_eq!(st, 200, "{a}");
    assert!(a["room"]["e2ee"]["record"].is_string() && a["room"]["world"]["statement"].is_string(), "{a}");
    let room = a["room"]["room"].as_str().unwrap().to_string();
    let ga = hex::decode(a["grant"].as_str().unwrap()).unwrap();
    let before = a["room"]["e2ee"]["log"].as_u64().unwrap();
    let ca = session(&server, with_grant(&server, &ga, &alice)).await;
    let _ta = WorldTrack::publish(&ca, &room, &alice, 40).await.expect("the world track");
    let mut grew = false;
    for _ in 0..100 {
        let (_, r) = call(addr, "GET", "/room", None, "10.0.1.9").await;
        if r["e2ee"]["log"].as_u64().unwrap_or(0) > before {
            grew = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    assert!(grew, "the clock's join of the actor, and its Add, are on the room's log");
    // A private world and a watch room are not one room.
    let bad = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/x\"\nrelay_url = \"ws://x/moq\"\n[co]\naudience_input = 2\n[world]\nmodel_id = \"{MODEL}\"\nwidth = 64\nheight = 64\nfps = 24\n")).unwrap();
    assert!(bad.settings().is_err(), "a world room is not also a room with co-performance");
    let zero = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/x\"\nrelay_url = \"ws://x/moq\"\n[world]\nmodel_id = \"{MODEL}\"\nwidth = 64\nheight = 64\nfps = 0\n")).unwrap();
    assert!(zero.settings().is_err(), "a world whose frame rate is 0 makes no statement that reads");
}