conclave-cli 0.3.4

//! Integration tests for the central server, driving simulated bridge clients over an in-memory
//! duplex against a real [`Hub`] + embedded store (DESIGN.md §17). These cover the M2 UATs; each
//! test name is prefixed with its UAT group (`server_register` / `server_auth` / `server_channel`
//! / `server_fanout` / `server_presence`) so the PRD's `test(/…/)` filters select them.

// Tests relax `unwrap_used` (house convention; DESIGN.md §22).
#![allow(clippy::unwrap_used)]

use std::{sync::Arc, time::Duration};

use pretty_assertions::assert_eq;
use tokio::io::DuplexStream;

use crate::{
    base::{Constant, SessionPath, Visibility},
    identity::Identity,
    protocol::{AdminOp, ProtocolError, ProtocolMessage, ProtocolRead as _, ProtocolWrite as _},
};

use super::{hub::Hub, session::handle_connection};

// -----------------------------------------------------------------------------
// Harness.
// -----------------------------------------------------------------------------

async fn hub_with(admins: &[(&str, Option<&str>)]) -> Arc<Hub> {
    let store = crate::store::Store::open_in_memory().await.unwrap();
    Hub::new(store, admins.iter().map(|(user, pin)| ((*user).to_owned(), pin.map(str::to_owned))).collect()).await.unwrap()
}

async fn hub() -> Arc<Hub> {
    hub_with(&[]).await
}

/// A simulated bridge client speaking the wire protocol over an in-memory duplex to a spawned
/// session driver.
struct Client {
    stream: DuplexStream,
}

impl Client {
    fn connect(hub: &Arc<Hub>) -> Self {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        tokio::spawn(handle_connection(Arc::clone(hub), server_end));
        Self { stream: client_end }
    }

    /// Sends a frame; a closed peer is tolerated (the test asserts on what it *receives*).
    async fn send(&mut self, frame: ProtocolMessage) {
        let _ = self.stream.send_message(&frame).await;
    }

    async fn recv(&mut self) -> ProtocolMessage {
        self.stream.recv_message().await.unwrap()
    }

    /// Receives the next frame, or `None` if nothing arrives within a short window (for asserting
    /// that a message did *not* reach this client).
    async fn try_recv(&mut self) -> Option<ProtocolMessage> {
        tokio::time::timeout(Duration::from_millis(200), self.stream.recv_message()).await.ok().and_then(Result::ok)
    }

    /// Claims a fresh username + machine and proves possession; returns the server's final frame.
    async fn register(&mut self, id: &Identity, username: &str, machine: &str, session: &str) -> ProtocolMessage {
        self.hello(session).await;
        let nonce = self.challenge().await;
        let pubkey = id.public_key().to_vec();
        let signature = id.sign(&nonce).unwrap().to_vec();
        self.send(ProtocolMessage::Register {
            username: username.to_owned(),
            machine: machine.to_owned(),
            pubkey: pubkey.clone(),
        })
        .await;
        self.send(ProtocolMessage::Auth { pubkey, signature }).await;
        self.recv_auth_result().await
    }

    /// Authenticates an already-enrolled key under a session handle; returns the server's response.
    async fn authenticate(&mut self, id: &Identity, session: &str) -> ProtocolMessage {
        self.hello(session).await;
        let nonce = self.challenge().await;
        let signature = id.sign(&nonce).unwrap().to_vec();
        self.send(ProtocolMessage::Auth {
            pubkey: id.public_key().to_vec(),
            signature,
        })
        .await;
        self.recv_auth_result().await
    }

    /// Reads the post-`Auth` frame, consuming the trailing `ServerInfo` on a successful `Established`.
    async fn recv_auth_result(&mut self) -> ProtocolMessage {
        let response = self.recv().await;
        if matches!(response, ProtocolMessage::Established { .. }) {
            let _ = self.recv().await;
        }
        response
    }

    async fn hello(&mut self, session: &str) {
        self.send(ProtocolMessage::Hello {
            protocol_version: Constant::PROTOCOL_VERSION,
            session: session.to_owned(),
        })
        .await;
    }

    async fn challenge(&mut self) -> Vec<u8> {
        match self.recv().await {
            ProtocolMessage::Challenge { nonce } => nonce,
            other => panic!("expected Challenge, got {other:?}"),
        }
    }

    async fn admin(&mut self, op: AdminOp) -> ProtocolMessage {
        self.send(ProtocolMessage::Admin(op)).await;
        self.recv().await
    }

    async fn join(&mut self, channel: &str, token: Option<&str>) -> ProtocolMessage {
        self.send(ProtocolMessage::Join {
            channel: channel.to_owned(),
            token: token.map(str::to_owned),
        })
        .await;
        self.recv().await
    }
}

fn established_path(frame: ProtocolMessage) -> SessionPath {
    match frame {
        ProtocolMessage::Established { path } => path,
        other => panic!("expected Established, got {other:?}"),
    }
}

fn invite_token(frame: ProtocolMessage) -> String {
    match frame {
        ProtocolMessage::InviteToken { token } => token,
        other => panic!("expected InviteToken, got {other:?}"),
    }
}

fn sorted_channel_names(frame: ProtocolMessage) -> Vec<String> {
    match frame {
        ProtocolMessage::ChannelList { channels } => {
            let mut names: Vec<String> = channels.into_iter().map(|c| c.name).collect();
            names.sort();
            names
        }
        other => panic!("expected ChannelList, got {other:?}"),
    }
}

fn is_unauthorized(frame: &ProtocolMessage) -> bool {
    matches!(frame, ProtocolMessage::Error(ProtocolError::Unauthorized(_)))
}

fn is_not_found(frame: &ProtocolMessage) -> bool {
    matches!(frame, ProtocolMessage::Error(ProtocolError::NotFound(_)))
}

// -----------------------------------------------------------------------------
// uat-001 — registration + enrollment; duplicate username rejected, stale live handle superseded.
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_register_claims_username_and_enrolls_machine() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut client = Client::connect(&hub);
    let path = established_path(client.register(&id, "aaron", "workstation", "razel").await);

    assert_eq!(path, SessionPath::new("aaron", "workstation", "razel"));
    assert!(hub.is_present(&path));

    // The key is now enrolled: a second connection authenticates with it and resolves the same
    // (user, machine).
    let mut again = Client::connect(&hub);
    let resolved = established_path(again.authenticate(&id, "dotagent").await);
    assert_eq!(resolved, SessionPath::new("aaron", "workstation", "dotagent"));
}

#[tokio::test]
async fn server_register_rejects_duplicate_username() {
    let hub = hub().await;
    let first = Identity::generate().unwrap();
    let second = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&first, "aaron", "workstation", "s1").await);

    // A different machine key trying to claim the same username is refused.
    let mut b = Client::connect(&hub);
    let response = b.register(&second, "aaron", "laptop", "s2").await;
    assert!(is_unauthorized(&response), "duplicate username must be rejected, got {response:?}");
}

#[tokio::test]
async fn server_reconnect_supersedes_a_stale_live_handle() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&id, "aaron", "workstation", "razel").await);

    // A fresh authenticated session for the same path takes over immediately — a half-open reconnect
    // must not wait out the idle reaper (#16). The new session is established...
    let mut b = Client::connect(&hub);
    let path = established_path(b.authenticate(&id, "razel").await);
    assert_eq!(path, SessionPath::new("aaron", "workstation", "razel"));

    // ...and the superseded session is force-dropped with a *self-describing* reason — a generic
    // "session terminated" reads like an auth failure and hides the mechanism (PRD-0012 T-002).
    match a.try_recv().await {
        Some(ProtocolMessage::Error(err)) => {
            assert!(err.to_string().contains("superseded"), "the termination frame must say the session was superseded, got: {err}");
        }
        None => panic!("the superseded session must receive a termination frame before its transport closes"),
        other => panic!("the superseded session must be terminated, got {other:?}"),
    }
}

// -----------------------------------------------------------------------------
// uat-002 — challenge-response resolves (user, machine); bad / revoked key refused.
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_auth_resolves_enrolled_key_to_user_and_machine() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut enroller = Client::connect(&hub);
    established_path(enroller.register(&id, "aaron", "workstation", "s1").await);

    let mut session = Client::connect(&hub);
    let path = established_path(session.authenticate(&id, "s2").await);
    assert_eq!(path.user, "aaron");
    assert_eq!(path.machine, "workstation");
    assert_eq!(path.session, "s2");
}

#[tokio::test]
async fn server_auth_refuses_unknown_key() {
    let hub = hub().await;
    let stranger = Identity::generate().unwrap();

    // A valid signature over the real nonce, but the key was never enrolled.
    let mut client = Client::connect(&hub);
    let response = client.authenticate(&stranger, "s").await;
    assert!(is_unauthorized(&response), "an unknown key must be refused, got {response:?}");
}

#[tokio::test]
async fn server_auth_refuses_bad_signature() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut enroller = Client::connect(&hub);
    established_path(enroller.register(&id, "aaron", "workstation", "s1").await);

    // Enrolled key, but the signature is over the wrong message.
    let mut client = Client::connect(&hub);
    client.hello("s2").await;
    let _nonce = client.challenge().await;
    let bad_signature = id.sign(&[0_u8; Constant::CHALLENGE_SIZE]).unwrap().to_vec();
    client
        .send(ProtocolMessage::Auth {
            pubkey: id.public_key().to_vec(),
            signature: bad_signature,
        })
        .await;
    let response = client.recv().await;
    assert!(is_unauthorized(&response), "a bad signature must be refused, got {response:?}");
}

// A self-enrolled machine key (AdminOp::MachineAdd) grants access only to whoever holds its private
// key: possession is proven at first-connect auth, so enrolling a key confers nothing on its own
// (PRD-0007 T-001, #8).
#[tokio::test]
async fn server_machine_add_key_still_proves_possession_at_auth() {
    let hub = hub().await;
    let workstation = Identity::generate().unwrap();
    let laptop = Identity::generate().unwrap();
    let impostor = Identity::generate().unwrap();

    // aaron registers workstation, then self-enrolls the laptop's public key on his own account.
    let mut aaron = Client::connect(&hub);
    established_path(aaron.register(&workstation, "aaron", "workstation", "razel").await);
    let ack = aaron
        .admin(AdminOp::MachineAdd {
            name: "laptop".to_owned(),
            pubkey: laptop.public_key().to_vec(),
        })
        .await;
    assert!(matches!(ack, ProtocolMessage::Ack { .. }), "machine add should be acked, got {ack:?}");

    // The enrolled key authenticates as aaron/laptop — possession proven by the challenge signature.
    let mut good = Client::connect(&hub);
    let path = established_path(good.authenticate(&laptop, "onthego").await);
    assert_eq!(path, SessionPath::new("aaron", "laptop", "onthego"));

    // An impostor key that was never enrolled is refused — enrollment alone grants nothing.
    let mut evil = Client::connect(&hub);
    assert!(is_unauthorized(&evil.authenticate(&impostor, "onthego").await), "an unenrolled key must not authenticate",);
}

#[tokio::test]
async fn server_auth_refuses_revoked_key() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut client = Client::connect(&hub);
    established_path(client.register(&id, "aaron", "workstation", "s1").await);

    // The user revokes their own machine (the lost-laptop kill switch, §5.1).
    let ack = client.admin(AdminOp::MachineRemove { name: "workstation".to_owned() }).await;
    assert!(matches!(ack, ProtocolMessage::Ack { .. }), "expected an ack for MachineRemove, got {ack:?}");

    // Reconnecting with the revoked key is now refused — the record is gone.
    let mut reconnect = Client::connect(&hub);
    let response = reconnect.authenticate(&id, "s2").await;
    assert!(is_unauthorized(&response), "a revoked key must be refused, got {response:?}");
}

// -----------------------------------------------------------------------------
// PRD-0007 T-001 (register_proof) — possession is proven before any durable write.
// An aborted registration handshake (a failed or mismatched Auth after Register) must
// persist nothing: no squatted username, no enrolled key (DESIGN.md §5.1).
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_register_proof_bad_signature_persists_nothing() {
    let hub = hub().await;
    let attacker = Identity::generate().unwrap();

    // An unauthenticated client claims the victim's username but fails the possession proof.
    let mut evil = Client::connect(&hub);
    evil.hello("s1").await;
    let _nonce = evil.challenge().await;
    let attacker_key = attacker.public_key().to_vec();
    evil.send(ProtocolMessage::Register {
        username: "aaron".to_owned(),
        machine: "evil".to_owned(),
        pubkey: attacker_key.clone(),
    })
    .await;
    let bad_signature = attacker.sign(&[0_u8; Constant::CHALLENGE_SIZE]).unwrap().to_vec();
    evil.send(ProtocolMessage::Auth {
        pubkey: attacker_key,
        signature: bad_signature,
    })
    .await;
    assert!(is_unauthorized(&evil.recv().await), "a failed possession proof must be rejected");

    // Nothing was persisted: the rightful owner can still claim the username with their own key...
    let owner = Identity::generate().unwrap();
    let mut legit = Client::connect(&hub);
    let result = legit.register(&owner, "aaron", "workstation", "s2").await;
    assert!(
        matches!(result, ProtocolMessage::Established { .. }),
        "an aborted handshake must not squat the username; got {result:?}"
    );

    // ...and the attacker's unproven key was never enrolled, so it cannot authenticate.
    let mut replay = Client::connect(&hub);
    assert!(is_unauthorized(&replay.authenticate(&attacker, "s3").await), "an unproven key must not have been enrolled");
}

#[tokio::test]
async fn server_register_proof_key_mismatch_persists_nothing() {
    let hub = hub().await;
    let claimed = Identity::generate().unwrap();
    let other = Identity::generate().unwrap();

    // Register one key, then prove possession of a *different* key (validly signed).
    let mut evil = Client::connect(&hub);
    evil.hello("s1").await;
    let nonce = evil.challenge().await;
    evil.send(ProtocolMessage::Register {
        username: "aaron".to_owned(),
        machine: "evil".to_owned(),
        pubkey: claimed.public_key().to_vec(),
    })
    .await;
    let signature = other.sign(&nonce).unwrap().to_vec();
    evil.send(ProtocolMessage::Auth {
        pubkey: other.public_key().to_vec(),
        signature,
    })
    .await;
    assert!(is_unauthorized(&evil.recv().await), "an auth key that does not match the registered key must be rejected");

    // The username is still free and neither key was enrolled.
    let owner = Identity::generate().unwrap();
    let mut legit = Client::connect(&hub);
    let result = legit.register(&owner, "aaron", "workstation", "s2").await;
    assert!(matches!(result, ProtocolMessage::Established { .. }), "a mismatched-key handshake must persist nothing; got {result:?}");
}

// -----------------------------------------------------------------------------
// PRD-0007 T-002 (admin_bootstrap) — a pinned admin username is claimable only by its bound
// key, so a fresh-deploy admin name cannot be squatted first-come (DESIGN.md §7).
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_admin_bootstrap_reserved_name_rejects_wrong_key() {
    let admin_id = Identity::generate().unwrap();
    let attacker = Identity::generate().unwrap();
    let bound = admin_id.public_key_base64();
    let hub = hub_with(&[("aaron", Some(&bound))]).await;

    // A different key cannot claim the pinned admin username, even with a valid possession proof.
    let mut evil = Client::connect(&hub);
    let denied = evil.register(&attacker, "aaron", "evil", "s1").await;
    assert!(is_unauthorized(&denied), "a pinned admin name must reject a non-matching key; got {denied:?}");

    // The bound key claims it and is granted admin (ListUsers is a server-admin-only op).
    let mut legit = Client::connect(&hub);
    established_path(legit.register(&admin_id, "aaron", "workstation", "s2").await);
    legit.send(ProtocolMessage::ListUsers).await;
    assert!(matches!(legit.recv().await, ProtocolMessage::UserList { .. }), "the key-bound admin must be granted admin rights");
}

// -----------------------------------------------------------------------------
// PRD-0007 T-006 (identity_validation) — usernames, machine names, and session handles must not
// be empty or contain the `/` path separator, or the `{user}/{machine}/{session}` path becomes
// ambiguous and `from`-attribution can be spoofed (finding #9).
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_register_rejects_a_username_with_a_path_separator() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();
    let mut client = Client::connect(&hub);
    let response = client.register(&id, "aa/ron", "workstation", "s1").await;
    assert!(matches!(response, ProtocolMessage::Error(_)), "a username with a path separator must be rejected, got {response:?}");
}

#[tokio::test]
async fn server_identity_validation_rejects_a_session_handle_with_a_path_separator() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();
    let mut client = Client::connect(&hub);
    let response = client.register(&id, "aaron", "workstation", "s/1").await;
    assert!(
        matches!(response, ProtocolMessage::Error(_)),
        "a session handle with a path separator must be rejected, got {response:?}"
    );
}

// -----------------------------------------------------------------------------
// PRD-0007 T-007 (visibility_oracle) — who() must not leak the existence of a private/unlisted
// channel to a non-member (#12), and must work for a participant of a public channel (#13).
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_who_visibility_oracle_hides_private_channel_existence() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let outsider = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "ops".to_owned(),
        visibility: Visibility::Private,
    })
    .await;

    let mut b = Client::connect(&hub);
    established_path(b.register(&outsider, "david", "wd", "sb").await);

    // who() on a private channel the caller cannot see must be indistinguishable from a channel
    // that does not exist at all.
    b.send(ProtocolMessage::Who { channel: Some("ops".to_owned()) }).await;
    let on_private = b.recv().await;
    b.send(ProtocolMessage::Who { channel: Some("ghost".to_owned()) }).await;
    let on_absent = b.recv().await;
    assert!(
        is_not_found(&on_private) && is_not_found(&on_absent),
        "who() must not leak private-channel existence, got {on_private:?} vs {on_absent:?}"
    );
}

#[tokio::test]
async fn server_who_is_open_to_a_public_channel_participant() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let guest = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "lobby".to_owned(),
        visibility: Visibility::Public,
    })
    .await;

    let mut b = Client::connect(&hub);
    let bpath = established_path(b.register(&guest, "david", "wd", "sb").await);
    assert!(matches!(b.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    // A participant of a public channel holds no ACL entry, but must still be able to see presence.
    b.send(ProtocolMessage::Who { channel: Some("lobby".to_owned()) }).await;
    match b.recv().await {
        ProtocolMessage::Presence { channel, sessions } => {
            assert_eq!(channel.as_deref(), Some("lobby"));
            assert!(sessions.contains(&bpath), "the participant must see themselves in presence");
        }
        other => panic!("expected Presence, got {other:?}"),
    }
}

// -----------------------------------------------------------------------------
// PRD-0007 T-008 (handshake_timeout) — an unauthenticated connection that never completes the
// handshake is dropped, so a silent pre-auth peer cannot hold a connection open forever (#15).
// -----------------------------------------------------------------------------

#[tokio::test(start_paused = true)]
async fn server_handshake_timeout_drops_a_silent_pre_auth_connection() {
    let hub = hub().await;
    let mut client = Client::connect(&hub);

    // Send nothing, then advance past the handshake deadline.
    tokio::time::advance(super::session::HANDSHAKE_TIMEOUT + Duration::from_secs(1)).await;

    assert!(
        matches!(client.recv().await, ProtocolMessage::Error(_)),
        "a silent pre-auth connection must be dropped with a timeout error"
    );
}

// -----------------------------------------------------------------------------
// uat-003 — channel create + ACL + invite redeem; private names never leak.
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_channel_create_acl_and_invite_redeem() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let guest = Identity::generate().unwrap();
    let latecomer = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    let create = a
        .admin(AdminOp::CreateChannel {
            name: "ops".to_owned(),
            visibility: Visibility::Private,
        })
        .await;
    assert!(matches!(create, ProtocolMessage::Ack { .. }));

    // A non-member cannot join a private channel without an invite.
    let mut b = Client::connect(&hub);
    let bpath = established_path(b.register(&guest, "david", "wd", "sb").await);
    let denied = b.join("ops", None).await;
    assert!(is_not_found(&denied), "private join without a token must be denied as not-found, got {denied:?}");

    // The channel admin mints a single-use invite; redeeming it joins and adds david to the ACL.
    let token = invite_token(
        a.admin(AdminOp::InviteCreate {
            channel: "ops".to_owned(),
            uses: Some(1),
            expires_in_secs: None,
        })
        .await,
    );
    let joined = b.join("ops", Some(&token)).await;
    assert!(matches!(joined, ProtocolMessage::Joined { .. }), "redeeming a valid invite must join, got {joined:?}");
    assert!(hub.subscribers("ops").contains(&bpath));

    // The single-use token is now spent — a second redeemer is refused.
    let mut c = Client::connect(&hub);
    established_path(c.register(&latecomer, "carol", "wc", "sc").await);
    let spent = c.join("ops", Some(&token)).await;
    assert!(is_not_found(&spent), "a spent single-use invite must be refused as not-found, got {spent:?}");
}

#[tokio::test]
async fn server_channel_acl_add_grants_membership() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let guest = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "ops".to_owned(),
        visibility: Visibility::Private,
    })
    .await;

    // Adding david to the ACL lets him join the private channel with no token.
    let acl = a
        .admin(AdminOp::AclAdd {
            channel: "ops".to_owned(),
            user: "david".to_owned(),
        })
        .await;
    assert!(matches!(acl, ProtocolMessage::Ack { .. }));

    let mut b = Client::connect(&hub);
    let bpath = established_path(b.register(&guest, "david", "wd", "sb").await);
    let joined = b.join("ops", None).await;
    assert!(matches!(joined, ProtocolMessage::Joined { .. }), "an ACL member must join without a token, got {joined:?}");
    assert!(hub.subscribers("ops").contains(&bpath));
}

#[tokio::test]
async fn server_channel_private_names_do_not_leak() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let outsider = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "lobby".to_owned(),
        visibility: Visibility::Public,
    })
    .await;
    a.admin(AdminOp::CreateChannel {
        name: "ops".to_owned(),
        visibility: Visibility::Private,
    })
    .await;

    // A non-member sees only the public channel; the private name never appears.
    let mut b = Client::connect(&hub);
    established_path(b.register(&outsider, "david", "wd", "sb").await);
    b.send(ProtocolMessage::ListChannels).await;
    assert_eq!(sorted_channel_names(b.recv().await), vec!["lobby".to_owned()]);

    // The owner (a member of both) sees both.
    a.send(ProtocolMessage::ListChannels).await;
    assert_eq!(sorted_channel_names(a.recv().await), vec!["lobby".to_owned(), "ops".to_owned()]);
}

// -----------------------------------------------------------------------------
// uat-002 (M5) — visibility tiers end-to-end.
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_visibility_unlisted_is_joinable_by_name_but_not_listed() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let outsider = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "secret-link".to_owned(),
        visibility: Visibility::Unlisted,
    })
    .await;

    // A non-member who knows the exact name can join an unlisted channel...
    let mut b = Client::connect(&hub);
    let bpath = established_path(b.register(&outsider, "david", "wd", "sb").await);
    assert!(matches!(b.join("secret-link", None).await, ProtocolMessage::Joined { .. }), "unlisted must be joinable by name");
    assert!(hub.subscribers("secret-link").contains(&bpath));

    // ...but it never appears in discovery.
    b.send(ProtocolMessage::ListChannels).await;
    assert_eq!(sorted_channel_names(b.recv().await), Vec::<String>::new(), "unlisted must not be listed in discovery");
}

#[tokio::test]
async fn server_visibility_private_is_hidden_and_gated() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let outsider = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "ops".to_owned(),
        visibility: Visibility::Private,
    })
    .await;

    let mut b = Client::connect(&hub);
    established_path(b.register(&outsider, "david", "wd", "sb").await);

    // Private is not joinable without an ACL entry or token, and is refused as not-found so its
    // existence never leaks (finding #12)...
    assert!(is_not_found(&b.join("ops", None).await), "private join without a token must be denied as not-found");
    // ...and never appears in discovery for a non-member.
    b.send(ProtocolMessage::ListChannels).await;
    assert_eq!(sorted_channel_names(b.recv().await), Vec::<String>::new(), "private must not leak into discovery");
}

// -----------------------------------------------------------------------------
// uat-004 — fan-out: channel message to all subscribers; whisper to exactly one.
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_fanout_channel_message_reaches_all_subscribers() {
    let hub = hub().await;
    let ida = Identity::generate().unwrap();
    let idb = Identity::generate().unwrap();
    let idc = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    let pa = established_path(a.register(&ida, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "lobby".to_owned(),
        visibility: Visibility::Public,
    })
    .await;
    assert!(matches!(a.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    let mut b = Client::connect(&hub);
    established_path(b.register(&idb, "david", "wd", "sb").await);
    assert!(matches!(b.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    let mut c = Client::connect(&hub);
    established_path(c.register(&idc, "carol", "wc", "sc").await);
    assert!(matches!(c.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    // A posts; the message reaches the other two subscribers, stamped with A's path.
    a.send(ProtocolMessage::ChannelMsg {
        channel: "lobby".to_owned(),
        from: pa.clone(),
        payload: crate::protocol::Payload::Plain("hello, agents".to_owned()),
    })
    .await;

    for listener in [&mut b, &mut c] {
        match listener.recv().await {
            ProtocolMessage::ChannelMsg { channel, from, payload } => {
                assert_eq!(channel, "lobby");
                assert_eq!(from, pa);
                assert_eq!(payload, crate::protocol::Payload::Plain("hello, agents".to_owned()));
            }
            other => panic!("expected a fanned-out ChannelMsg, got {other:?}"),
        }
    }
}

// -----------------------------------------------------------------------------
// PRD-0013 — retained history: read-since after a gap; visibility-uniform refusal.
// -----------------------------------------------------------------------------

#[tokio::test]
async fn server_history_read_since_returns_missed_messages_in_order() {
    let hub = hub().await;
    let ida = Identity::generate().unwrap();
    let idb = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    let pa = established_path(a.register(&ida, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "lobby".to_owned(),
        visibility: Visibility::Public,
    })
    .await;
    assert!(matches!(a.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    // A posts while B is entirely offline — the uat-001 gap.
    for body in ["one", "two"] {
        a.send(ProtocolMessage::ChannelMsg {
            channel: "lobby".to_owned(),
            from: pa.clone(),
            payload: crate::protocol::Payload::Plain(body.to_owned()),
        })
        .await;
        assert!(matches!(a.recv().await, ProtocolMessage::Ack { .. }));
    }

    // B arrives later, joins, and catches up from zero: both messages, oldest-first, stamped.
    let mut b = Client::connect(&hub);
    established_path(b.register(&idb, "david", "wd", "sb").await);
    assert!(matches!(b.join("lobby", None).await, ProtocolMessage::Joined { .. }));
    b.send(ProtocolMessage::ReadSince { channel: "lobby".to_owned(), since_ms: 0 }).await;
    let (first_ts, second_ts) = match b.recv().await {
        ProtocolMessage::History { channel, messages } => {
            assert_eq!(channel, "lobby");
            let bodies: Vec<_> = messages
                .iter()
                .map(|m| match &m.payload {
                    crate::protocol::Payload::Plain(text) => text.clone(),
                    other @ crate::protocol::Payload::Encrypted(_) => panic!("expected a plain payload, got {other:?}"),
                })
                .collect();
            assert_eq!(bodies, ["one", "two"], "history must be complete and oldest-first");
            assert!(messages.iter().all(|m| m.from == pa), "history must carry the server-stamped sender path");
            (messages[0].ts_ms, messages[1].ts_ms)
        }
        other => panic!("expected a History page, got {other:?}"),
    };
    assert!(first_ts <= second_ts, "server stamps must be monotonic-ish");

    // The watermark is exclusive: asking since the first message returns only the second.
    b.send(ProtocolMessage::ReadSince {
        channel: "lobby".to_owned(),
        since_ms: first_ts,
    })
    .await;
    match b.recv().await {
        ProtocolMessage::History { messages, .. } => {
            assert_eq!(messages.len(), 1, "the watermark message itself must not be re-delivered");
            assert_eq!(messages[0].ts_ms, second_ts);
        }
        other => panic!("expected a History page, got {other:?}"),
    }
}

#[tokio::test]
async fn server_history_is_refused_visibility_uniformly() {
    let hub = hub().await;
    let ida = Identity::generate().unwrap();
    let idb = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&ida, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "secret".to_owned(),
        visibility: Visibility::Private,
    })
    .await;

    // A non-member reading a private channel and reading a nonexistent channel must be
    // indistinguishable (PRD-0007 visibility uniformity, inherited by history).
    let mut b = Client::connect(&hub);
    established_path(b.register(&idb, "david", "wd", "sb").await);
    b.send(ProtocolMessage::ReadSince {
        channel: "secret".to_owned(),
        since_ms: 0,
    })
    .await;
    let private_refusal = match b.recv().await {
        ProtocolMessage::Error(err) => err.to_string().replace("secret", "<c>"),
        other => panic!("a non-member history read must be refused, got {other:?}"),
    };
    b.send(ProtocolMessage::ReadSince {
        channel: "no-such-channel".to_owned(),
        since_ms: 0,
    })
    .await;
    let missing_refusal = match b.recv().await {
        ProtocolMessage::Error(err) => err.to_string().replace("no-such-channel", "<c>"),
        other => panic!("a history read on a missing channel must be refused, got {other:?}"),
    };
    assert_eq!(private_refusal, missing_refusal, "history refusals must not leak channel existence");
}

#[tokio::test]
async fn server_fanout_whisper_reaches_exactly_one_session() {
    let hub = hub().await;
    let ida = Identity::generate().unwrap();
    let idb = Identity::generate().unwrap();
    let idc = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    let pa = established_path(a.register(&ida, "aaron", "wa", "sa").await);
    let mut b = Client::connect(&hub);
    let pb = established_path(b.register(&idb, "david", "wd", "sb").await);
    let mut c = Client::connect(&hub);
    established_path(c.register(&idc, "carol", "wc", "sc").await);

    // A whispers to B's exact path.
    a.send(ProtocolMessage::Whisper {
        from: pa.clone(),
        target: pb.clone(),
        payload: crate::protocol::Payload::Plain("just for you".to_owned()),
    })
    .await;

    match b.recv().await {
        ProtocolMessage::Whisper { from, target, payload } => {
            assert_eq!(from, pa);
            assert_eq!(target, pb);
            assert_eq!(payload, crate::protocol::Payload::Plain("just for you".to_owned()));
        }
        other => panic!("expected a Whisper, got {other:?}"),
    }

    // The successful whisper is acked back to the sender (PRD-0008 T-001).
    assert!(matches!(a.recv().await, ProtocolMessage::Ack { .. }), "a successful whisper must be acked to the sender");

    // C — a third live session — receives nothing.
    assert!(c.try_recv().await.is_none(), "a whisper must not reach a third session");

    // Whispering to an offline / unknown path errors back to the sender.
    a.send(ProtocolMessage::Whisper {
        from: pa.clone(),
        target: SessionPath::new("ghost", "box", "sess"),
        payload: crate::protocol::Payload::Plain("anyone?".to_owned()),
    })
    .await;
    assert!(matches!(a.recv().await, ProtocolMessage::Error(ProtocolError::NotFound(_))));
}

// -----------------------------------------------------------------------------
// uat-005 — heartbeat reaps a half-open connection; revocation force-drops sessions.
// -----------------------------------------------------------------------------

#[tokio::test(start_paused = true)]
async fn server_presence_reaps_half_open_connection() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut client = Client::connect(&hub);
    let path = established_path(client.register(&id, "aaron", "workstation", "razel").await);
    assert!(hub.is_present(&path));

    // Idle well past the timeout with no keepalive → the reaper drops the zombie connection.
    tokio::time::advance(Duration::from_secs(120)).await;
    assert_eq!(hub.reap_idle(Duration::from_secs(60)), 1);
    assert!(!hub.is_present(&path), "a half-open connection must be reaped");
}

#[tokio::test(start_paused = true)]
async fn server_presence_ping_keeps_a_session_alive() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut client = Client::connect(&hub);
    let path = established_path(client.register(&id, "aaron", "workstation", "razel").await);

    // A keepalive refreshes the heartbeat, so a subsequent reap with the same timeout spares it.
    tokio::time::advance(Duration::from_secs(50)).await;
    client.send(ProtocolMessage::Ping).await;
    assert!(matches!(client.recv().await, ProtocolMessage::Pong));

    tokio::time::advance(Duration::from_secs(50)).await;
    assert_eq!(hub.reap_idle(Duration::from_secs(60)), 0);
    assert!(hub.is_present(&path), "a heartbeated session must not be reaped");
}

#[tokio::test]
async fn server_presence_machine_remove_force_drops_sessions() {
    let hub = hub().await;
    let id = Identity::generate().unwrap();

    let mut c1 = Client::connect(&hub);
    let p1 = established_path(c1.register(&id, "aaron", "workstation", "s1").await);
    let mut c2 = Client::connect(&hub);
    let p2 = established_path(c2.authenticate(&id, "s2").await);
    assert!(hub.is_present(&p1) && hub.is_present(&p2));

    // Revoking the machine force-drops every live session for that (user, machine) immediately.
    let ack = c1.admin(AdminOp::MachineRemove { name: "workstation".to_owned() }).await;
    assert!(matches!(ack, ProtocolMessage::Ack { .. }));
    assert!(!hub.is_present(&p1));
    assert!(!hub.is_present(&p2));
}

#[tokio::test]
async fn server_presence_kick_removes_from_channel() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let guest = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    let pa = established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "lobby".to_owned(),
        visibility: Visibility::Public,
    })
    .await;
    assert!(matches!(a.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    let mut b = Client::connect(&hub);
    let pb = established_path(b.register(&guest, "david", "wd", "sb").await);
    assert!(matches!(b.join("lobby", None).await, ProtocolMessage::Joined { .. }));
    assert_eq!(hub.subscribers("lobby").len(), 2);

    // The channel admin kicks david; he leaves the channel but stays connected.
    let ack = a
        .admin(AdminOp::Kick {
            channel: "lobby".to_owned(),
            target: "david".to_owned(),
        })
        .await;
    assert!(matches!(ack, ProtocolMessage::Ack { .. }));

    let subscribers = hub.subscribers("lobby");
    assert!(subscribers.contains(&pa));
    assert!(!subscribers.contains(&pb), "the kicked session must be dropped from the channel");
    assert!(hub.is_present(&pb), "a channel kick must not disconnect the session");
}

#[tokio::test]
async fn server_presence_ban_drops_from_channel_and_blocks_rejoin() {
    let hub = hub().await;
    let owner = Identity::generate().unwrap();
    let guest = Identity::generate().unwrap();

    let mut a = Client::connect(&hub);
    established_path(a.register(&owner, "aaron", "wa", "sa").await);
    a.admin(AdminOp::CreateChannel {
        name: "lobby".to_owned(),
        visibility: Visibility::Public,
    })
    .await;
    assert!(matches!(a.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    let mut b = Client::connect(&hub);
    let pb = established_path(b.register(&guest, "david", "wd", "sb").await);
    assert!(matches!(b.join("lobby", None).await, ProtocolMessage::Joined { .. }));

    // Banning david removes him from the channel and blocks him from re-joining, even though the
    // channel is public.
    let ack = a
        .admin(AdminOp::Ban {
            channel: "lobby".to_owned(),
            user: "david".to_owned(),
        })
        .await;
    assert!(matches!(ack, ProtocolMessage::Ack { .. }));
    assert!(!hub.subscribers("lobby").contains(&pb), "a ban must drop the session from the channel");

    let rejoin = b.join("lobby", None).await;
    assert!(is_unauthorized(&rejoin), "a banned user must not be able to rejoin, got {rejoin:?}");
}