//! Calls kept by the room service (WAI live-messaging §10, `[call]`) against a relay on
//! loopback that trusts its authority:
//!
//! - a call between two principals: the caller proves its key and places the call (the
//!   authority signs its descriptor and the caller's grant), rings the callee with an
//!   invitation it signs; the callee reads its inbox on a proof of its key, verifies the
//!   invitation and its call offline, and accepts; both enter with their grants and proofs,
//!   publish audio and video, and each receives the other's; the clock records each `join` and
//!   `publish`; an invitation whose signature does not verify, and one ringing a principal the
//!   call does not invite, are refused;
//! - a call among three: one invitee declines, and the other's invitation is missed, signed by
//!   the room's clock at its deadline and found in its inbox; a late accept is refused;
//! - every call record verifies offline, as §10 verifies a call.

use ed25519_dalek::SigningKey;
use joule_moq::draft22::data::{FetchEntry, SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
use joule_moq::draft22::Parameters;
use joule_moq::grant_proof::{self as gp, GrantProof};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, 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::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
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_messaging::{self as lm, answer, Answer, CallInput, Invitation, State};

const STEP: Duration = Duration::from_secs(20);

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

fn pk(k: &SigningKey) -> [u8; 32] {
    k.verifying_key().to_bytes()
}

fn now_us() -> u64 {
    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64
}

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 },
    ];
    c.receipts.door = vec![];
    RelayServer::start(c).await.expect("the relay starts")
}

async fn http(addr: std::net::SocketAddr, method: &str, path: &str, headers: &[(&str, String)], body: Option<&Value>) -> (u16, BTreeMap<String, String>, Value) {
    let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
    let b = body.map(|v| v.to_string()).unwrap_or_default();
    let extra: String = headers.iter().map(|(k, v)| format!("{k}: {v}\r\n")).collect();
    let req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\nCF-Connecting-IP: 10.0.0.1\r\n{extra}Content-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();
    let hs = head.lines().skip(1).filter_map(|l| l.split_once(':')).map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())).collect();
    (status, hs, serde_json::from_str(body).unwrap_or(Value::Null))
}

/// A principal's requests: a proof naming its grant, or none (live-rooms §10.5.3).
struct Who {
    key: SigningKey,
    grant: Option<Vec<u8>>,
    challenge: Option<[u8; 32]>,
}

impl Who {
    fn new(k: SigningKey) -> Who {
        Who { key: k, grant: None, challenge: None }
    }

    async fn req(&mut self, addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>) -> (u16, Value) {
        for _ in 0..2 {
            let ch = match self.challenge {
                Some(c) => c,
                None => {
                    let (_, _, v) = http(addr, "GET", "/call/challenge", &[], None).await;
                    hex::decode(v["challenge"].as_str().unwrap()).unwrap().try_into().unwrap()
                }
            };
            // `/call/ring` is the caller's grant's; every other request a proof of the key alone.
            let grant = self.grant.as_ref().filter(|_| path == "/call/ring");
            let digest = grant.map_or([0; 32], |g| gp::grant_digest_and_principal(g).unwrap().0);
            let proof = GrantProof::sign(digest, &self.key, ch, &gp::context(method, "room.test", path)).unwrap();
            let mut hs = vec![(gp::HEADER_PROOF, gp::base64url(&proof.to_bytes()))];
            if let Some(g) = grant {
                hs.push(("Authorization", format!("Bearer {}", gp::base64url(g))));
            }
            let (st, h, v) = http(addr, method, path, &hs, body).await;
            self.challenge = h.get(&gp::HEADER_CHALLENGE.to_ascii_lowercase()).and_then(|c| gp::from_base64url(c)).and_then(|c| c.try_into().ok());
            if st != 401 {
                return (st, v);
            }
        }
        panic!("{method} {path}: refused twice for its proof");
    }
}

async fn session(server: &RelayServer, grant: &[u8], k: &SigningKey) -> 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())?)) };
    let opts = ClientOptions { grants: vec![(grant.to_vec(), k.clone())], target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path), ..ClientOptions::default() };
    // 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 with a grant and its proof")
}

/// Publish `k`'s audio and video tracks in `room` and write one object on each: what the
/// call carries.
async fn publish_media(c: &Client, room: &str, k: &SigningKey) -> Vec<(joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)> {
    let mut held = Vec::new();
    for (alias, m) in (10..).zip(["audio", "video"]) {
        let ns = TrackNamespace::from_strs([room, hex::encode(pk(k)).as_str(), m]);
        let mut p = Properties::new();
        p.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 64));
        held.push(c.publish(ns, TrackName::from_str("0"), alias, Parameters::new(), p).await.expect("PUBLISH"));
        let header = SubgroupHeader { track_alias: alias, group_id: 0, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: false, first_object: true };
        let mut b = Vec::new();
        header.encode(&mut b);
        let mut codec = SubgroupCodec::new(header);
        codec.encode_object(&SubgroupObject::normal(0, format!("{m} of {}", hex::encode(&pk(k)[..4])).into_bytes()), &mut b).unwrap();
        let mut s = c.conn.open_uni().await.unwrap();
        s.write_all(&b).await.unwrap();
        let _ = s.finish();
    }
    held
}

/// The objects of `k`'s track of media `m` the relay serves, by FETCH.
async fn received(c: &Client, room: &str, k: &SigningKey, m: &str) -> Vec<Vec<u8>> {
    let ns = TrackNamespace::from_strs([room, hex::encode(pk(k)).as_str(), m]);
    for _ in 0..100 {
        if let Ok((_, entries)) = c.fetch(ns.clone(), TrackName::from_str("0"), Parameters::new()).await {
            let got: Vec<Vec<u8>> = entries.into_iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some(o.payload) } else { None }).collect();
            if !got.is_empty() {
                return got;
            }
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    panic!("no {m} of {} reached the relay", hex::encode(&pk(k)[..4]));
}

fn kinds_of(chain: &[Vec<u8>]) -> Vec<(String, [u8; 32])> {
    chain
        .iter()
        .filter_map(|r| {
            let ev = wai::live_events::Event::parse(r).ok()?;
            ev.verify_signature().ok()?;
            let wai::live_events::Payload::Lifecycle { subject, .. } = wai::live_events::parse_payload(ev.kind, &ev.payload, 1024).ok()? else { return None };
            Some((wai::live_events::kind_name(ev.kind)?.to_string(), subject))
        })
        .collect()
}

async fn chain(c: &Client, room: &str, clock: &str, t0_us: u64) -> Vec<Vec<u8>> {
    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 = BTreeMap::new();
    for e in entries {
        let FetchEntry::Object(o) = e else { continue };
        let arr: Vec<Value> = serde_json::from_slice(&o.payload).unwrap_or_default();
        for r in arr {
            if let Ok(rec) = wai::live_events::from_msf_record(&r.to_string(), t0_us) {
                let ev = wai::live_events::Event::parse(&rec).unwrap();
                by_seq.insert(ev.seq, rec);
            }
        }
    }
    by_seq.into_values().collect()
}

/// The call as `GET /call` gives it, verified offline as §10 verifies a call: each invitation's
/// callee and state.
fn verified(v: &Value) -> Vec<([u8; 32], State)> {
    let h = |x: &Value| hex::decode(x.as_str().unwrap()).unwrap();
    let list = |k: &str| v[k].as_array().unwrap().iter().map(h).collect::<Vec<_>>();
    let c = lm::verify_call(&CallInput { descriptor: h(&v["descriptor"]), call: h(&v["call"]), invitations: list("invitations"), answers: list("answers") }).expect("the call verifies");
    c.rung.iter().map(|r| (r.invitation.callee, r.state)).collect()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_call_rings_is_answered_carries_media_and_records_a_missed_call() {
    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[call]\nring_ms = 1500\n[rate]\nclient_burst = 64\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();
    svc.start();
    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 clock_hex = hex::encode(keys.clock.verifying_key().to_bytes());

    // A call room's grants come with its call.
    let (st, _, _) = http(addr, "POST", "/grant", &[], Some(&json!({"principal": hex::encode([1u8; 32]), "role": "publisher"}))).await;
    assert_eq!(st, 403);

    // Alice calls Bob.
    let (alice, bob) = (key(0x71), key(0x72));
    let mut ca = Who::new(alice.clone());
    let (st, p) = ca.req(addr, "POST", "/call/place", Some(&json!({"invitees": [hex::encode(pk(&bob))]}))).await;
    assert_eq!(st, 200, "{p}");
    ca.grant = Some(hex::decode(p["grant"].as_str().unwrap()).unwrap());
    let room = p["room"]["room"].as_str().unwrap().to_string();
    let room_id: [u8; 32] = hex::decode(&room).unwrap().try_into().unwrap();
    let t0 = p["room"]["t0_us"].as_u64().unwrap();
    let call_digest: [u8; 32] = hex::decode(p["digest"].as_str().unwrap()).unwrap().try_into().unwrap();
    let (st, busy) = Who::new(key(0x79)).req(addr, "POST", "/call/place", Some(&json!({"invitees": [hex::encode(pk(&bob))]}))).await;
    assert_eq!((st, busy["error"].as_str()), (409, Some("busy")), "one call at a time");

    // Ringing: an invitation whose signature does not verify, and one ringing a principal the
    // call does not invite, are refused; Alice's own rings Bob.
    let t = now_us() - t0;
    let mut forged = Invitation::new(&alice, room_id, call_digest, pk(&bob), t);
    forged.sig[5] ^= 1;
    let (st, v) = ca.req(addr, "POST", "/call/ring", Some(&json!({"invitations": [hex::encode(forged.to_bytes())]}))).await;
    assert_eq!((st, v["error"].as_str()), (409, Some("bad_signature")), "{v}");
    let stranger = Invitation::new(&alice, room_id, call_digest, pk(&key(0x7f)), t);
    let (st, v) = ca.req(addr, "POST", "/call/ring", Some(&json!({"invitations": [hex::encode(stranger.to_bytes())]}))).await;
    assert_eq!((st, v["error"].as_str()), (409, Some("callee_not_invited")));
    let inv = Invitation::new(&alice, room_id, call_digest, pk(&bob), t);
    let (st, v) = ca.req(addr, "POST", "/call/ring", Some(&json!({"invitations": [hex::encode(inv.to_bytes())]}))).await;
    assert_eq!(st, 200, "{v}");

    // Bob's inbox holds it; he verifies it and its call before his device rings, then accepts.
    let mut cb = Who::new(bob.clone());
    let (st, ib) = cb.req(addr, "GET", "/inbox", None).await;
    assert_eq!(st, 200);
    let item = &ib["invitations"][0];
    assert_eq!(item["state"].as_str(), Some("ringing"));
    let h = |x: &Value| hex::decode(x.as_str().unwrap()).unwrap();
    let mut offline = lm::Call::new(&h(&item["descriptor"]), &h(&item["call"])).expect("the call descriptor verifies");
    let r = offline.offer_invitation(&h(&item["invitation"])).expect("the invitation verifies").clone();
    assert_eq!((r.invitation.caller, r.invitation.callee), (pk(&alice), pk(&bob)));
    let acc = Answer::new(&bob, room_id, inv.digest(), answer::ACCEPT, now_us() - t0);
    let (st, a) = cb.req(addr, "POST", "/call/answer", Some(&json!({"answer": hex::encode(acc.to_bytes())}))).await;
    assert_eq!((st, a["state"].as_str()), (200, Some("accepted")), "{a}");
    let bob_grant = hex::decode(a["grant"].as_str().expect("the callee's grant")).unwrap();

    // Both enter with their grants and proofs and publish audio and video; each receives the
    // other's.
    let sa = session(&server, ca.grant.as_ref().unwrap(), &alice).await;
    let sb = session(&server, &bob_grant, &bob).await;
    let ha = publish_media(&sa, &room, &alice).await;
    let hb = publish_media(&sb, &room, &bob).await;
    for (c, peer) in [(&sa, &bob), (&sb, &alice)] {
        for m in ["audio", "video"] {
            let got = received(c, &room, peer, m).await;
            assert_eq!(got, vec![format!("{m} of {}", hex::encode(&pk(peer)[..4])).into_bytes()], "{m} reaches the other party");
        }
    }
    // The clock records each party's join and its publishing.
    let mut life = Vec::new();
    for _ in 0..100 {
        life = kinds_of(&chain(&sa, &room, &clock_hex, t0).await);
        if life.len() >= 6 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    for p in [pk(&alice), pk(&bob)] {
        let k: Vec<&str> = life.iter().filter(|(_, s)| *s == p).map(|(k, _)| k.as_str()).collect();
        assert_eq!(k, ["join", "publish", "publish"], "{life:?}");
    }
    let (st, cv) = ca.req(addr, "GET", "/call", None).await;
    assert_eq!(st, 200);
    assert_eq!(verified(&cv), vec![(pk(&bob), State::Accepted)]);

    // The call ends as a room does; the next call is a new room. Alice calls Carol and Dave:
    // Carol declines; Dave never answers, and his invitation is missed at its deadline, signed by
    // the room's clock.
    drop(ha);
    drop(hb);
    sa.close(0, "done");
    sb.close(0, "done");
    let (carol, dave) = (key(0x73), key(0x74));
    let mut ca = Who::new(alice.clone());
    let mut placed = Value::Null;
    for _ in 0..100 {
        let (st, p) = ca.req(addr, "POST", "/call/place", Some(&json!({"invitees": [hex::encode(pk(&carol)), hex::encode(pk(&dave))]}))).await;
        if st == 200 {
            placed = p;
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    assert_ne!(placed["room"]["room"].as_str(), Some(room.as_str()), "a new room for the next call: {placed}");
    ca.grant = Some(hex::decode(placed["grant"].as_str().unwrap()).unwrap());
    let room2: [u8; 32] = hex::decode(placed["room"]["room"].as_str().unwrap()).unwrap().try_into().unwrap();
    let t02 = placed["room"]["t0_us"].as_u64().unwrap();
    let d2: [u8; 32] = hex::decode(placed["digest"].as_str().unwrap()).unwrap().try_into().unwrap();
    let t = now_us() - t02;
    let (ic, id) = (Invitation::new(&alice, room2, d2, pk(&carol), t), Invitation::new(&alice, room2, d2, pk(&dave), t));
    let (st, _) = ca.req(addr, "POST", "/call/ring", Some(&json!({"invitations": [hex::encode(ic.to_bytes()), hex::encode(id.to_bytes())]}))).await;
    assert_eq!(st, 200);
    let mut cc = Who::new(carol.clone());
    let dec = Answer::new(&carol, room2, ic.digest(), answer::DECLINE, now_us() - t02);
    let (st, v) = cc.req(addr, "POST", "/call/answer", Some(&json!({"answer": hex::encode(dec.to_bytes())}))).await;
    assert_eq!((st, v["state"].as_str()), (200, Some("declined")));
    assert!(v.get("grant").is_none(), "a decline gets no grant");
    // Carol cannot answer for Dave.
    let forged = Answer::new(&carol, room2, id.digest(), answer::ACCEPT, now_us() - t02);
    let (st, _) = cc.req(addr, "POST", "/call/answer", Some(&json!({"answer": hex::encode(forged.to_bytes())}))).await;
    assert_eq!(st, 409, "an accept signed by another than the callee");
    tokio::time::sleep(Duration::from_millis(2_200)).await;
    let mut cd = Who::new(dave.clone());
    let (_, ib) = cd.req(addr, "GET", "/inbox", None).await;
    let item = &ib["invitations"][0];
    assert_eq!(item["state"].as_str(), Some("missed"), "{ib}");
    let missed = Answer::read(&h(&item["answer"])).unwrap();
    assert_eq!((missed.kind, missed.signer), (answer::MISSED, keys.clock.verifying_key().to_bytes()), "the clock signs the missed call");
    let late = Answer::new(&dave, room2, id.digest(), answer::ACCEPT, now_us() - t02);
    let (st, v) = cd.req(addr, "POST", "/call/answer", Some(&json!({"answer": hex::encode(late.to_bytes())}))).await;
    assert_eq!((st, v["error"].as_str()), (409, Some("already_answered")));
    let (_, cv) = cc.req(addr, "GET", "/call", None).await;
    assert_eq!(verified(&cv), vec![(pk(&carol), State::Declined), (pk(&dave), State::Missed)]);
    // A party to neither call reads neither.
    let (st, _) = Who::new(key(0x7e)).req(addr, "GET", "/call", None).await;
    assert_eq!(st, 403);
    server.shutdown().await;
}