//! An offer a room presents, through the service and the relay (WAI shoppable-events):
//!
//! - a room with co-performance, whose first publisher is its lead and holds the direct power;
//! - `POST /offer` from a viewer's grant is refused (403), an offer with its stock written zero
//!   when it is unknown is refused with the extension's own reason, as is one whose description
//!   is another item's, and nothing is signed for either;
//! - the lead's offer is signed by the clock: the description as a `meta` event and the offer as
//!   an anchored `meta` event of the clock's chain;
//! - a subscriber of the clock's events track reads both from the relay, holds the description,
//!   reads the offer as shoppable-events §5.6 reads it with the room keys admitted, and a sink's
//!   rule presents it inside its window and not after;
//! - the same record offered with only a guest's key admitted is refused, and with one bit of
//!   its payload changed its signature fails.

use ed25519_dalek::SigningKey;
use joule_moq::draft22::data::FetchEntry;
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, FloorConfig, 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_events::{anchor_track, kind, Event};
use wai::shoppable_events as shop;

const STEP: Duration = Duration::from_secs(20);
const S: u64 = 1_000_000;

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.clone(), "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: true },
    ];
    c.floor = Some(FloorConfig { room_keys: vec![clock], ..FloorConfig::default() });
    RelayServer::start(c).await.expect("the relay starts")
}

/// One HTTP/1.1 request: its status, its headers (lowercase names) and its JSON.
async fn call(addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>, headers: &[(&str, String)]) -> (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 mut req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\nCF-Connecting-IP: 10.7.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n", b.len());
    for (k, v) in headers {
        req.push_str(&format!("{k}: {v}\r\n"));
    }
    req.push_str("\r\n");
    req.push_str(&b);
    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))
}

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

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

fn pubhex(k: &SigningKey) -> String {
    hex::encode(pk(k))
}

/// A page's identity in the room: its key and its grant.
struct Who {
    key: SigningKey,
    grant: Vec<u8>,
    challenge: Option<[u8; 32]>,
}

impl Who {
    /// A request with this grant and a proof over the last challenge the service gave it, asking
    /// for one first when it holds none.
    async fn ask(&mut self, addr: std::net::SocketAddr, path: &str, body: &Value) -> (u16, Value) {
        for _ in 0..2 {
            let mut headers = vec![("Authorization", format!("Bearer {}", gp::base64url(&self.grant)))];
            if let Some(c) = self.challenge.take() {
                let (digest, _) = gp::grant_digest_and_principal(&self.grant).unwrap();
                let p = GrantProof::sign(digest, &self.key, c, &gp::context("POST", "room.test", path)).unwrap();
                headers.push((gp::HEADER_PROOF, gp::base64url(&p.to_bytes())));
            }
            let (st, hs, v) = call(addr, "POST", path, Some(body), &headers).await;
            self.challenge = hs.get(&gp::HEADER_CHALLENGE.to_ascii_lowercase()).and_then(|c| gp::from_base64url(c)).and_then(|b| b.try_into().ok());
            if st != 401 {
                return (st, v);
            }
        }
        (401, Value::Null)
    }
}

async fn session(server: &RelayServer, who: &Who) -> 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![(who.grant.clone(), who.key.clone())], target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path), ..ClientOptions::default() };
    Client::connect_proving(dial, opts).await.expect("SETUP")
}

/// Every record of a principal's events track as the relay serves it, by `seq`.
async fn chain(c: &Client, room: &str, who: &str, t0: u64) -> Vec<Vec<u8>> {
    let ns = TrackNamespace::from_strs([room, who, "events"]);
    let Ok((_, entries)) = c.fetch(ns, TrackName::from_str("0"), Parameters::new()).await else { return Vec::new() };
    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();
        for r in arr {
            let rec = wai::live_events::from_msf_record(&r.to_string(), t0).unwrap();
            by_seq.insert(Event::parse(&rec).unwrap().seq, rec);
        }
    }
    by_seq.into_values().collect()
}

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

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_room_presents_an_offer_its_subscribers_read() {
    timeout(Duration::from_secs(180), 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 = 3\njoin_wait_s = 30\n[rate]\nclient_burst = 40\n[co]\naudience_input = 2\nlatency_budget_ms = 1500\nlead = \"first\"\nai = false\nfidelity_every_s = 0\ncommit_after_ms = 800\nlead_ms = 200\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");

    // The lead (the room's first publisher) and a viewer.
    let grant = |k: SigningKey, role: &'static str| async move {
        let (st, _, g) = call(addr, "POST", "/grant", Some(&json!({"principal": pubhex(&k), "role": role})), &[]).await;
        assert_eq!(st, 200, "{g}");
        (Who { key: k, grant: hex::decode(g["grant"].as_str().unwrap()).unwrap(), challenge: None }, g)
    };
    let (lead_k, viewer_k) = (key(0x31), key(0x33));
    let (mut lead, gl) = grant(lead_k.clone(), "publisher").await;
    let (mut viewer, _) = grant(viewer_k.clone(), "viewer").await;
    let room = gl["room"]["room"].as_str().unwrap().to_string();
    let room_id: [u8; 32] = hex::decode(&room).unwrap().try_into().unwrap();
    let t0 = gl["room"]["t0_us"].as_u64().unwrap();
    let rc = now_us() - t0;

    // The merchant's item, its description, and an offer open for a minute from now, anchored to
    // an object of the lead's video.
    let merchant = key(0x51);
    let item = shop::Item::build(&merchant, "sku/lamp-01", "Desk lamp, brass", "Example Lighting Co.", "https://shop.example/wai/intent");
    let offer = |amount: u64, stock: Option<u64>, description: [u8; 32]| shop::Offer {
        merchant: pk(&merchant),
        item: "sku/lamp-01".into(),
        description,
        amount,
        currency: "XTS".into(),
        exponent: 2,
        start: rc,
        end: rc + 60 * S,
        show: shop::Show::Everywhere,
        regions: vec![],
        stock,
    };
    let track = anchor_track(&format!("{room}-{}-video", pubhex(&lead_k)), "0");
    let anchor = json!({"track": hex::encode(track), "group": 7, "object": 0});
    let good = offer(1250, Some(40), shop::digest_of(&item));
    let request = |o: &Value| json!({"offer": o, "item": item, "anchor": anchor});

    // Refused, and nothing signed: a viewer's grant; an unknown stock written as zero; a
    // description that is another item's; no anchor.
    let (st, v) = viewer.ask(addr, "/offer", &request(&good.to_value())).await;
    assert_eq!((st, v["error"].as_str()), (403, Some("power")), "{v}");
    let mut zero = good.to_value();
    zero["stock"] = json!({"stated": false, "count": 0});
    let (st, v) = lead.ask(addr, "/offer", &request(&zero)).await;
    assert_eq!((st, v["reason"].as_str()), (400, Some(shop::STOCK_ZERO_WHEN_UNKNOWN)), "{v}");
    let (st, v) = lead.ask(addr, "/offer", &request(&offer(1250, None, [7; 32]).to_value())).await;
    assert_eq!((st, v["reason"].as_str()), (400, Some(shop::ITEM_MISMATCH)), "{v}");
    let (st, v) = lead.ask(addr, "/offer", &json!({"offer": good.to_value(), "item": item, "anchor": {"track": "00", "group": 7, "object": 0}})).await;
    assert_eq!((st, v["reason"].as_str()), (400, Some(shop::NOT_ANCHORED)), "{v}");

    // The lead's offer: the clock signs the description and the offer.
    let (st, v) = lead.ask(addr, "/offer", &request(&good.to_value())).await;
    assert_eq!(st, 200, "{v}");
    let id = v["offer"].as_str().unwrap().to_string();
    let record = gp::from_base64url(v["record"].as_str().unwrap()).unwrap();
    assert_eq!(hex::encode(Event::parse(&record).unwrap().digest()), id);

    // A subscriber reads the clock's events track from the relay.
    let clock = hex::encode(keys.clock.verifying_key().to_bytes());
    let watcher = session(&server, &viewer).await;
    let mut metas: Vec<Vec<u8>> = Vec::new();
    for _ in 0..100 {
        metas = chain(&watcher, &room, &clock, t0).await.into_iter().filter(|r| Event::parse(r).unwrap().kind == kind::META).collect();
        if metas.iter().any(|r| Event::parse(r).unwrap().payload.starts_with(shop::OFFER_MAGIC)) {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    let offers: Vec<&Vec<u8>> = metas.iter().filter(|r| Event::parse(r).unwrap().payload.starts_with(shop::OFFER_MAGIC)).collect();
    assert_eq!(offers.len(), 1, "one offer on the clock's track: the refused requests signed nothing");
    assert_eq!(offers[0], &record, "the record the relay serves is the one the service answered with");
    // The description, held from the room: the payload after `WOI1`.
    let descriptions: Vec<String> =
        metas.iter().map(|r| Event::parse(r).unwrap().payload).filter(|p| p.starts_with(shop::ITEM_MAGIC)).map(|p| String::from_utf8(p[4..].to_vec()).unwrap()).collect();
    assert_eq!(descriptions.len(), 1);
    let room_keys = [keys.authority.verifying_key().to_bytes(), keys.clock.verifying_key().to_bytes()];
    let read = shop::read_offer(&room_id, offers[0], &room_keys, Some(&descriptions[0])).expect("the offer reads, with the description the room carried");
    assert_eq!(hex::encode(read.id), id);
    assert_eq!(read.offer, good);
    assert_eq!(read.offer.price_text(), "12.50 XTS");
    assert_eq!(read.event.anchor.map(|a| (a.track, a.group, a.object)), Some((track, 7, 0)));
    // The offer is one event of the clock's chain, after the description.
    let ev = Event::parse(offers[0]).unwrap();
    let before = metas.iter().map(|r| Event::parse(r).unwrap()).find(|e| e.payload.starts_with(shop::ITEM_MAGIC)).unwrap();
    assert!(before.seq < ev.seq && ev.principal == room_keys[1]);

    // What a sink decides: presented in its window, not after it, and not without the description.
    let nowhere = shop::aggregate(&read.offer.regions, None);
    assert_eq!(shop::presentable(&read.offer, true, rc + S, &nowhere), Ok(()));
    assert_eq!(shop::presentable(&read.offer, true, rc + 60 * S, &nowhere), Err(shop::OUTSIDE_WINDOW));
    assert_eq!(shop::presentable(&read.offer, false, rc + S, &nowhere), Err(shop::ITEM_NOT_HELD));

    // The same record where the clock is not admitted, and with one bit of its price changed.
    assert_eq!(shop::read_offer(&room_id, offers[0], &[pk(&viewer_k)], None).err(), Some(shop::AUTHOR_NOT_ADMITTED));
    let mut changed = offers[0].clone();
    let at = changed.windows(13).position(|w| w == b"\"amount\":1250").expect("the price in the payload") + 12;
    changed[at] = b'1';
    assert_eq!(shop::read_offer(&room_id, &changed, &room_keys, None).err(), Some(shop::OFFER_SIGNATURE_INVALID));
    drop(watcher);
}