//! A conversation kept by the room service (WAI live-messaging, `[messaging]`), its members
//! keying through a real MLS group (`joule-moq-mls`) and sealing with `joule-moq`'s secure
//! objects, as pages do:
//!
//! - three members join, key through the service's log and message one another; each opens
//!   what the others seal, and `wai-rs` verifies every message as §11 does;
//! - one goes offline; the others go on, the service restarts from its history store, a
//!   fourth member joins and a member other than the lowest commits its Add (§7.2); the one
//!   offline returns and catches up from the store: every message of both epochs opens;
//! - a thread, reactions, an edit and a retraction, verified, and the retracted message kept
//!   in verification, stated retracted;
//! - acknowledgements of delivery and reading taken, and one of another member refused;
//! - a member removed: the next commit excludes it, it opens nothing sealed after its
//!   removal, the store serves it nothing past its removal (`not_a_member`) and takes nothing
//!   more from it;
//! - every record in the log verifies offline, and the store holds no plaintext.
//!
//! And `wai-rs`'s sealing of a message object (confidentiality §4, as live-messaging §4.2 uses
//! it) and `joule-moq`'s secure objects open what the other seals.

use ed25519_dalek::SigningKey;
use joule_moq::draft22::kvp::Properties;
use joule_moq::grant_proof::{self as gp, GrantProof};
use joule_moq::secure_objects::room::{Incoming, Keyring, Outgoing};
use joule_moq::secure_objects::{Plaintext, Suite};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_mls::{CommitRule, Received, RoomMember};
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, Config, Service};
use serde_json::{json, Value};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;
use wai::live_messaging::{self as lm, kind, Conversation, MessageEntry, Open, Outcome};

const STEP: Duration = Duration::from_secs(20);
const STAMP: &str = "2026-10-26T00:00:00Z";

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

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

fn b64(b: &[u8]) -> String {
    gp::base64url(b)
}

/// One HTTP/1.1 request: the status, the headers (names in lowercase) and the JSON.
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 making requests: its key, its grant (none before it joins), and the challenge
/// the service gave it last (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 }
    }

    /// A request with this principal's grant (or none: a proof naming no grant) and a proof
    /// over the last challenge, asked again once with a fresh challenge when refused 401.
    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", "/msg/challenge", &[], None).await;
                    hex::decode(v["challenge"].as_str().unwrap()).unwrap().try_into().unwrap()
                }
            };
            let digest = self.grant.as_ref().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, b64(&proof.to_bytes()))];
            if let Some(g) = &self.grant {
                hs.push(("Authorization", format!("Bearer {}", b64(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");
    }
}

/// A member as a page is one: its requests, its MLS state, its place in the log, and the
/// conversation as it verifies it.
struct M {
    who: Who,
    m: RoomMember,
    at: u64,
    view: Conversation,
    room: [u8; 32],
    t0_us: u64,
    seen: Vec<Received>,
    /// The original payloads it opened, by position.
    opened: BTreeMap<u64, Vec<u8>>,
}

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

fn entry_of(v: &Value) -> MessageEntry {
    let h = |k: &str| hex::decode(v[k].as_str().unwrap()).unwrap();
    MessageEntry { author: h("author").try_into().unwrap(), group: v["group"].as_u64().unwrap(), properties: h("properties"), payload: h("payload"), receipts: h("receipts") }
}

fn ns_of(room: &[u8; 32], author: &[u8; 32]) -> TrackNamespace {
    TrackNamespace::from_strs([hex::encode(room).as_str(), hex::encode(author).as_str(), "messages"])
}

impl M {
    async fn join(addr: std::net::SocketAddr, k: SigningKey) -> M {
        let mut who = Who::new(k.clone());
        let (st, v) = who.req(addr, "POST", "/msg/join", Some(&json!({}))).await;
        assert_eq!(st, 200, "{v}");
        who.grant = Some(hex::decode(v["grant"].as_str().unwrap()).unwrap());
        let c = &v["conversation"];
        let h = |k: &str| hex::decode(c[k].as_str().unwrap()).unwrap();
        let room: [u8; 32] = h("room").try_into().unwrap();
        let clock: [u8; 32] = h("clock").try_into().unwrap();
        let view = Conversation::new(&h("descriptor"), Some(&h("record")), &h("statement")).expect("the conversation verifies");
        let mut m = RoomMember::new(room, clock, k, Suite::AesGcm128Sha256);
        m.set_commit_rule(CommitRule::AnyMember);
        let mut me = M { who, m, at: 0, view, room, t0_us: c["t0_us"].as_u64().unwrap(), seen: vec![], opened: BTreeMap::new() };
        let kp = me.m.key_package().unwrap();
        let (st, v) = me.who.req(addr, "POST", "/msg/key-package", Some(&json!({"key_package": hex::encode(kp)}))).await;
        assert_eq!(st, 200, "{v}");
        me
    }

    fn me(&self) -> [u8; 32] {
        pk(&self.who.key)
    }

    /// Read the log to its end and act, until nothing is left to read or do. Its status:
    /// 200, or the refusal of a read.
    async fn pump(&mut self, addr: std::net::SocketAddr) -> u16 {
        loop {
            let (st, v) = self.who.req(addr, "GET", &format!("/msg/log?from={}&wait_ms=0", self.at), None).await;
            if st != 200 {
                return st;
            }
            let entries = v["entries"].as_array().unwrap().clone();
            for e in &entries {
                let seq = e["seq"].as_u64().unwrap();
                assert!(seq >= self.at, "the log goes forward");
                if let Some(hx) = e["event"].as_str() {
                    let rec = hex::decode(hx).unwrap();
                    let ev = wai::live_events::Event::parse(&rec).unwrap();
                    ev.verify_signature().expect("the clock's record verifies");
                    if let Ok(wai::live_events::Payload::Lifecycle { subject, .. }) = wai::live_events::parse_payload(ev.kind, &ev.payload, 1024) {
                        match wai::live_events::kind_name(ev.kind) {
                            Some("join") => self.m.event(&joule_moq_mls::Event::Join { subject, keyed: true }),
                            Some("leave") => self.m.event(&joule_moq_mls::Event::Leave { subject, reason: 1 }),
                            _ => {}
                        }
                    }
                } else if let Some(hx) = e["mls"].as_str() {
                    let r = self.m.receive(&hex::decode(hx).unwrap()).expect("every carried message applies");
                    self.seen.push(r);
                } else if let Some(hx) = e["epoch"].as_str() {
                    let b = hex::decode(hx).unwrap();
                    self.view.offer_epoch(seq, &b).expect("the clock's epoch statement verifies");
                    // §7.3: a member checks the statement against the epoch its group reached.
                    let s = lm::EpochStatement::read(&b).unwrap();
                    if let Some(r) = self.m.keyring().roster(s.epoch).filter(|_| self.m.epoch() == Some(s.epoch)) {
                        let mine: Vec<(u32, [u8; 32])> = r.iter().map(|(l, p)| (*l, *p)).collect();
                        assert_eq!(mine, s.roster, "the epoch statement names the roster the group reached");
                    }
                } else if let Some(mv) = e.get("message") {
                    let me = entry_of(mv);
                    let ns = ns_of(&self.room, &me.author);
                    let name = TrackName::from_str("0");
                    let props = Properties::decode(&me.properties, "properties").unwrap();
                    let open = self.m.open(&Incoming { ns: &ns, name: &name, group: me.group, object: 0, priority: lm::PRIORITY, properties: &props, payload: &me.payload });
                    match open {
                        Ok(o) => {
                            self.opened.insert(seq, o.plaintext.payload.clone());
                            self.view.offer_message(seq, &me, Open::Opened(&o.plaintext.payload));
                        }
                        Err(_) => {
                            self.view.offer_message(seq, &me, Open::No);
                        }
                    }
                } else if let Some(hx) = e["ack"].as_str() {
                    let _ = self.view.offer_ack(seq, &hex::decode(hx).unwrap());
                }
                self.at = seq + 1;
            }
            self.at = self.at.max(v["next"].as_u64().unwrap());
            let mut acted = false;
            if self.m.owes_create() {
                let info = self.m.stage_create().unwrap();
                if self.post_mls(addr, &[info]).await != 200 {
                    self.m.withdraw();
                }
                acted = true;
            } else if !self.m.staged() && self.m.is_committer() && self.m.clock_pending() {
                let c = self.m.stage_commit(false).unwrap();
                let frames: Vec<Vec<u8>> = std::iter::once(c.commit).chain(c.welcome).collect();
                if self.post_mls(addr, &frames).await != 200 {
                    self.m.withdraw();
                }
                acted = true;
            }
            if entries.is_empty() && !acted {
                return 200;
            }
        }
    }

    async fn post_mls(&mut self, addr: std::net::SocketAddr, frames: &[Vec<u8>]) -> u16 {
        self.who.req(addr, "POST", "/msg/mls", Some(&json!({"frames": frames.iter().map(hex::encode).collect::<Vec<_>>()}))).await.0
    }

    /// Seal and post a message of `kind` naming `target`, once the log is read to its end: its
    /// status, and its link hash.
    async fn send(&mut self, addr: std::net::SocketAddr, k: u8, target: [u8; 32], body: &str) -> (u16, Value, [u8; 32]) {
        let (post, link) = self.prepare(addr, k, target, body).await;
        let (st, v) = self.who.req(addr, "POST", "/msg/post", Some(&post)).await;
        (st, v, link)
    }

    /// A message sealed and receipted, as `send` posts it: its post's body and its link hash.
    async fn prepare(&mut self, addr: std::net::SocketAddr, k: u8, target: [u8; 32], body: &str) -> (Value, [u8; 32]) {
        self.pump(addr).await;
        let (group, parent) = self.view.next_of(&self.me());
        let rec = lm::Message::new(&self.who.key, self.room, group, now_us().saturating_sub(self.t0_us), k, target, body.as_bytes());
        let ns = ns_of(&self.room, &self.me());
        let name = TrackName::from_str("0");
        let pt = Plaintext { payload: rec.to_bytes(), ..Default::default() };
        let s = self.m.seal(&Outgoing { ns: &ns, name: &name, group, object: 0, priority: lm::PRIORITY, plaintext: &pt, immutable: &[], mutable: &[] }).expect("an epoch to seal under");
        let (receipts, link) = lm::receipt(&self.who.key, group, &s.payload, parent, STAMP);
        let body = json!({"group": group, "properties": hex::encode(s.properties.to_bytes().unwrap()), "payload": hex::encode(&s.payload), "receipts": hex::encode(receipts)});
        (body, link)
    }

    async fn text(&mut self, addr: std::net::SocketAddr, k: u8, target: [u8; 32], body: &str) -> [u8; 32] {
        let (st, v, link) = self.send(addr, k, target, body).await;
        assert_eq!(st, 200, "{body}: {v}");
        link
    }
}

async fn start(cfg: &Config, keys: &Keys) -> (Arc<Service>, std::net::SocketAddr, tokio::task::JoinHandle<()>) {
    let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), cfg);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let task = tokio::spawn(http::serve(svc.clone(), listener));
    (svc, addr, task)
}

/// Pump every member in turn until the epochs are `want`.
async fn settle(addr: std::net::SocketAddr, ms: &mut [&mut M], want: &[Option<u64>]) {
    for _ in 0..100 {
        for m in ms.iter_mut() {
            m.pump(addr).await;
        }
        if ms.iter().map(|m| m.m.epoch()).collect::<Vec<_>>() == want {
            return;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    panic!("epochs {:?}, not {want:?}", ms.iter().map(|m| m.m.epoch()).collect::<Vec<_>>());
}

fn accepted(view: &Conversation, link: &[u8; 32]) -> lm::Accepted {
    view.by_link(link).and_then(|v| v.accepted.clone()).expect("an accepted message")
}

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

async fn run() {
    let keys = Keys::generate();
    let dir = std::env::temp_dir().join(format!("joule-moq-room-messaging-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    let cfg = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://127.0.0.1:1/moq\"\n[messaging]\nstore = \"{}\"\nmax_members = 8\nmax_text = 512\n[rate]\nclient_burst = 64\n", dir.display())).unwrap();
    let (svc, addr, task) = start(&cfg, &keys).await;

    // Alice, the first, owns the conversation and creates its group; Bob and Carol join, and
    // Alice commits their Adds.
    let mut a = M::join(addr, key(0x41)).await;
    settle(addr, &mut [&mut a], &[Some(0)]).await;
    let mut b = M::join(addr, key(0x42)).await;
    let mut c = M::join(addr, key(0x43)).await;
    settle(addr, &mut [&mut a, &mut b, &mut c], &[Some(1), Some(1), Some(1)]).await;
    assert!(a.seen.contains(&Received::Created));
    assert!(b.seen.contains(&Received::Joined(1)) && c.seen.contains(&Received::Joined(1)));

    // Three members message: each opens what the others seal.
    let hello = a.text(addr, kind::TEXT, [0; 32], "hello from alice").await;
    b.pump(addr).await;
    b.text(addr, kind::TEXT, [0; 32], "hello from bob").await;
    c.pump(addr).await;
    c.text(addr, kind::TEXT, [0; 32], "hello from carol").await;
    for m in [&mut a, &mut b, &mut c] {
        m.pump(addr).await;
        let texts: Vec<String> = m.view.messages.iter().filter_map(|v| v.accepted.as_ref()).map(|x| String::from_utf8(x.body.clone()).unwrap()).collect();
        assert_eq!(texts, ["hello from alice", "hello from bob", "hello from carol"], "every member opens every message");
    }

    // Carol goes offline. Alice starts a thread, Bob replies, Alice reacts.
    let root = a.text(addr, kind::TEXT, [0; 32], "lunch at noon?").await;
    b.pump(addr).await;
    let reply = b.text(addr, kind::REPLY, root, "yes").await;
    a.pump(addr).await;
    a.text(addr, kind::REACTION, reply, "+1").await;
    let (_, before) = a.who.req(addr, "GET", "/msg/log?from=0&wait_ms=0", None).await;

    // The service restarts from its history store.
    task.abort();
    drop(svc);
    tokio::time::sleep(Duration::from_millis(200)).await;
    let (_svc, addr, _task) = start(&cfg, &keys).await;
    for m in [&mut a, &mut b, &mut c] {
        m.who.challenge = None;
    }
    let (st, after) = a.who.req(addr, "GET", "/msg/log?from=0&wait_ms=0", None).await;
    assert_eq!(st, 200);
    assert_eq!(after["entries"], before["entries"], "the log is as it was, every entry at its position");

    // Alice edits her root; Alice goes away, Dave joins and Bob, not the lowest leaf, commits
    // his Add (§7.2).
    a.pump(addr).await;
    a.text(addr, kind::EDIT, root, "lunch at one?").await;
    let mut d = M::join(addr, key(0x44)).await;
    settle(addr, &mut [&mut b, &mut d], &[Some(2), Some(2)]).await;
    assert!(d.seen.contains(&Received::Joined(2)), "dave joined by bob's commit");
    let dave_hi = d.text(addr, kind::TEXT, [0; 32], "dave here").await;
    b.pump(addr).await;
    let oops = b.text(addr, kind::TEXT, [0; 32], "wrong conversation").await;
    b.text(addr, kind::RETRACTION, oops, "").await;
    a.pump(addr).await;
    assert_eq!(a.m.epoch(), Some(2));
    a.text(addr, kind::REPLY, root, "see you there").await;

    // Carol returns and catches up from the store: both epochs, before and after the restart.
    c.pump(addr).await;
    assert_eq!(c.m.epoch(), Some(2));
    let cv = &c.view;
    assert!(cv.messages.iter().all(|v| v.outcome == Outcome::Accepted), "{:?}", cv.messages.iter().map(|v| &v.outcome).collect::<Vec<_>>());
    let r = accepted(cv, &root);
    assert_eq!((r.text.as_deref(), r.edits, r.replies.len()), (Some(&b"lunch at one?"[..]), 1, 2), "the root edited, with its thread");
    assert_eq!(cv.by_link(&root).unwrap().status(), Some("edited"));
    let rp = accepted(cv, &reply);
    assert_eq!(rp.reactions.iter().map(|(w, body, _)| (*w, body.clone())).collect::<Vec<_>>(), vec![(a.me(), b"+1".to_vec())]);
    let x = cv.by_link(&oops).expect("a retracted message stays in verification");
    assert_eq!((x.status(), x.accepted.as_ref().unwrap().text.as_ref()), (Some("retracted"), None));
    assert!(accepted(cv, &dave_hi).text.is_some());
    assert_eq!(c.opened.len(), cv.messages.len(), "carol opened every message, sealed in epochs 1 and 2");

    // Acknowledgements: Carol's delivery and reading of Alice's track taken; one naming a
    // message it does not hold refused; one of another member's refused.
    let (a_next, a_link) = c.view.next_of(&a.me());
    let ack = lm::Ack::new(&c.who.key, c.room, a.me(), a_next - 1, a_link.unwrap(), lm::ack::READ, 1).to_bytes();
    let (st, v) = c.who.req(addr, "POST", "/msg/ack", Some(&json!({"ack": hex::encode(&ack)}))).await;
    assert_eq!(st, 200, "{v}");
    let bad = lm::Ack::new(&c.who.key, c.room, a.me(), a_next - 1, [9; 32], lm::ack::DELIVERED, 2).to_bytes();
    let (st, v) = c.who.req(addr, "POST", "/msg/ack", Some(&json!({"ack": hex::encode(&bad)}))).await;
    assert_eq!((st, v["error"].as_str()), (409, Some("ack_unknown")));
    let others = lm::Ack::new(&d.who.key, c.room, a.me(), a_next - 1, a_link.unwrap(), lm::ack::DELIVERED, 3).to_bytes();
    let (st, _) = c.who.req(addr, "POST", "/msg/ack", Some(&json!({"ack": hex::encode(&others)}))).await;
    assert_eq!(st, 403);
    a.pump(addr).await;
    assert_eq!(a.view.acked.get(&a.me()).and_then(|m| m.get(&c.me())).map(|x| (x.delivered, x.read)), Some((Some(a_next - 1), Some(a_next - 1))));

    // Bob seals a message in the epoch he is in, and holds it back.
    let (late, _) = b.prepare(addr, kind::TEXT, [0; 32], "sent after my removal").await;
    // Alice removes Bob; only an owner may. The next commit excludes him.
    let (st, _) = d.who.req(addr, "POST", "/msg/remove", Some(&json!({"principal": hex::encode(b.me())}))).await;
    assert_eq!(st, 403, "dave is not an owner");
    let (st, v) = a.who.req(addr, "POST", "/msg/remove", Some(&json!({"principal": hex::encode(b.me())}))).await;
    assert_eq!(st, 200, "{v}");
    settle(addr, &mut [&mut a, &mut c, &mut d], &[Some(3), Some(3), Some(3)]).await;
    assert_eq!(b.pump(addr).await, 403, "bob reads what he is owed up to his removal, then nothing: not_a_member");
    assert!(b.seen.contains(&Received::Removed(3)), "bob applies the commit that removes him");
    let removal_end = b.at;

    // Alice writes in the epoch Bob is not in.
    let after_removal = a.text(addr, kind::TEXT, [0; 32], "bob cannot read this").await;
    c.pump(addr).await;
    assert!(accepted(&c.view, &after_removal).text.is_some());
    // The store serves Bob nothing past his removal: from where he stopped, nothing more, and
    // from the end, not_a_member.
    let (st, v) = b.who.req(addr, "GET", &format!("/msg/log?from={removal_end}&wait_ms=0"), None).await;
    assert_eq!(st, 403, "{v}");
    assert_eq!(v["error"].as_str(), Some("not_a_member"));
    // He opens nothing sealed after it, though he holds the object.
    let pos = *c.opened.keys().last().unwrap();
    let (_, log) = c.who.req(addr, "GET", &format!("/msg/log?from={pos}&wait_ms=0"), None).await;
    let e = entry_of(&log["entries"][0]["message"]);
    let props = Properties::decode(&e.properties, "properties").unwrap();
    let ns = ns_of(&b.room, &e.author);
    let opened = b.m.open(&Incoming { ns: &ns, name: &TrackName::from_str("0"), group: e.group, object: 0, priority: lm::PRIORITY, properties: &props, payload: &e.payload });
    assert!(opened.is_err(), "a removed member opens nothing of a later epoch: {opened:?}");
    // The store takes nothing more from him, not even what he sealed before his removal: he is not
    // admitted, and the message is a removed member's (§11.3 step 2).
    let (st, v) = b.who.req(addr, "POST", "/msg/post", Some(&late)).await;
    assert_eq!((st, v["error"].as_str()), (403, Some("not_a_member")));
    let late_entry = MessageEntry {
        author: b.me(),
        group: late["group"].as_u64().unwrap(),
        properties: hex::decode(late["properties"].as_str().unwrap()).unwrap(),
        payload: hex::decode(late["payload"].as_str().unwrap()).unwrap(),
        receipts: hex::decode(late["receipts"].as_str().unwrap()).unwrap(),
    };
    c.pump(addr).await;
    assert_eq!(c.view.check_post(&late_entry).map(|_| ()), Err(lm::Reason::SenderNotMember), "a verifier refuses a removed member's message");
    // He is not admitted again.
    let mut again = Who::new(key(0x42));
    let (st, v) = again.req(addr, "POST", "/msg/join", Some(&json!({}))).await;
    assert_eq!((st, v["error"].as_str()), (403, Some("removed")));

    // Every record of the log verifies offline: the whole log as Alice is owed it, every
    // message opened with what Alice opened.
    a.pump(addr).await;
    let (_, full) = a.who.req(addr, "GET", "/msg/log?from=0&wait_ms=0", None).await;
    assert_eq!(full["withheld"].as_array().unwrap().len(), 0, "alice, a member throughout, is owed every entry");
    let (_, _, conv) = http(addr, "GET", "/msg", &[], None).await;
    let h = |k: &str| hex::decode(conv[k].as_str().unwrap()).unwrap();
    let mut offline = Conversation::new(&h("descriptor"), Some(&h("record")), &h("statement")).unwrap();
    let (mut events, mut messages, mut acks) = (0, 0, 0);
    for e in full["entries"].as_array().unwrap() {
        let seq = e["seq"].as_u64().unwrap();
        if let Some(x) = e["event"].as_str() {
            wai::live_events::Event::parse(&hex::decode(x).unwrap()).unwrap().verify_signature().unwrap();
            events += 1;
        } else if let Some(x) = e["epoch"].as_str() {
            offline.offer_epoch(seq, &hex::decode(x).unwrap()).unwrap();
        } else if let Some(mv) = e.get("message") {
            let p = a.opened.get(&seq).expect("alice opened it");
            assert_eq!(offline.offer_message(seq, &entry_of(mv), Open::Opened(p)), Outcome::Accepted);
            messages += 1;
        } else if let Some(x) = e["ack"].as_str() {
            offline.offer_ack(seq, &hex::decode(x).unwrap()).unwrap();
            acks += 1;
        }
    }
    assert_eq!((events, messages, acks), (5, 12, 1), "four joins and a leave, every message, one acknowledgement");
    assert_eq!(offline.epochs().len(), 4);

    // The store holds ciphertext and signed records only.
    let journal = std::fs::read(dir.join("journal")).unwrap();
    for plain in ["hello from alice", "lunch at", "wrong conversation", "bob cannot read this"] {
        assert!(!journal.windows(plain.len()).any(|w| w == plain.as_bytes()), "the store holds no plaintext: {plain}");
    }
    let _ = hello;
    let _ = std::fs::remove_dir_all(&dir);
}

/// `wai-rs`'s sealing of a message object and `joule-moq`'s secure objects agree: each opens
/// what the other seals, under one epoch's secret, at one KID.
#[test]
fn wai_rs_and_secure_objects_seal_alike() {
    let room = [7u8; 32];
    let alice = key(0x51);
    let secret = [0x5c; 32];
    let roster: joule_moq::secure_objects::room::Roster = [(0u32, pk(&key(0x50))), (1, pk(&alice))].into();
    let mut ring = Keyring::new(Suite::AesGcm128Sha256, pk(&alice));
    ring.enter(4, secret, roster.clone());
    let record = lm::Message::new(&alice, room, 3, 9, kind::TEXT, [0; 32], b"sealed by wai-rs").to_bytes();
    let kid = lm::seal::kid(4, 1);
    let (props, sealed) = lm::seal::seal(lm::SUITE_AES128, &secret, &room, &pk(&alice), 3, kid, &record, None);
    let ns = ns_of(&room, &pk(&alice));
    let name = TrackName::from_str("0");
    let p = Properties::decode(&props, "properties").unwrap();
    let mut reader = Keyring::new(Suite::AesGcm128Sha256, pk(&key(0x50)));
    reader.enter(4, secret, roster);
    let o = reader.open(&Incoming { ns: &ns, name: &name, group: 3, object: 0, priority: lm::PRIORITY, properties: &p, payload: &sealed }, None).expect("secure objects open what wai-rs sealed");
    assert_eq!(o.plaintext.payload, record);
    // And the other way.
    let pt = Plaintext { payload: record.clone(), padding: Some(24), ..Default::default() };
    let s = ring.seal(&Outgoing { ns: &ns, name: &name, group: 5, object: 0, priority: lm::PRIORITY, plaintext: &pt, immutable: &[], mutable: &[] }).unwrap().unwrap();
    assert_eq!(s.kid, kid);
    let got = lm::seal::open(lm::SUITE_AES128, &secret, &room, &pk(&alice), 5, &s.properties.to_bytes().unwrap(), &s.payload).expect("wai-rs opens what secure objects sealed");
    assert_eq!(got, record);
    assert!(lm::seal::open(lm::SUITE_AES128, &[0; 32], &room, &pk(&alice), 5, &s.properties.to_bytes().unwrap(), &s.payload).is_err(), "another secret opens nothing");
}