//! **IRCv3 `setname` propagation** — the `ENCAP SETNAME` carrier and the shared local
//! co-member fan-out (P11 slice 90).
//!
//! A realname (GECOS) change originates at the [`SETNAME`](crate::command::setname) command (a
//! local user) or arrives over a link as `:<uid> ENCAP * SETNAME :<realname>` (a remote user).
//! Either way the network sees the same line, and any server a changed user shares a channel
//! with notifies its own local `setname`-capable members.
//!
//! Three pieces, mirroring [`super::chghost`] but **without** the non-cap reconnect fallback —
//! the spec forbids sending `SETNAME` to clients that did not negotiate the capability, and a
//! realname change (unlike a host change) needs no client refresh:
//!
//! - [`local_setname`] — outbound: a local change broadcasts `:<uid> ENCAP * SETNAME
//!   :<realname>` to **every** peer (the source is us, so none is excluded).
//! - [`apply_encap_setname`] — inbound: a peer's `ENCAP SETNAME` updates the remote user's
//!   [`RemoteUser`](super::RemoteUser) mirror + its registry record and notifies local capable
//!   co-members. The onward split-horizon relay is done by the caller
//!   ([`super::forward::inbound_encap`]), exactly as every other `ENCAP` line relays.
//! - [`notify_co_members`] — the shared fan-out both paths use: deliver `:nick!user@host SETNAME
//!   :<realname>` to every **local** channel co-member of the changed user that negotiated
//!   `setname`, plus extended-monitor watchers.
//!
//! # Divergences (documented)
//!
//! - **Single-uplink no-op.** With one direct peer the outbound broadcast reaches zero or one
//!   peer — the standing [`super::relay`] divergence.
//! - **No `leveva-integration` differential** — the C 2.11 oracle has no `setname`/`ENCAP
//!   SETNAME` interpretation (its `ENCAP` is an opaque placeholder relay).

use crate::ident::Uid;
use crate::server::ServerContext;
use crate::Message;

use super::PeerLink;

/// Broadcast a locally-originated realname change to every peer as `:<uid> ENCAP * SETNAME
/// :<realname>`. A no-op when there are no peers.
pub fn local_setname(ctx: &ServerContext, from: &Uid, realname: &str) {
    if ctx.peers.is_empty() {
        return;
    }
    let wire = Message::builder("ENCAP")
        .prefix(from.as_str())
        .param("*")
        .param("SETNAME")
        .trailing(realname)
        .to_wire();
    ctx.peers.broadcast(None, wire);
}

/// Notify every **local** channel co-member of `uid` that its realname changed, as a single
/// `:nick!user@host SETNAME :<realname>` — but only to members that negotiated `setname` (the
/// spec forbids sending it to others, and there is no reconnect fallback). The changed user
/// itself is excluded (the command path self-echoes separately; the inbound path's user is
/// remote). Also pushes the line to extended-monitor watchers (extended-monitor + setname) that
/// share no channel with the changed user.
pub(crate) fn notify_co_members(
    ctx: &ServerContext,
    uid: &Uid,
    nick: &str,
    user: &str,
    host: &str,
    realname: &str,
) {
    let mask = format!("{nick}!{user}@{host}");
    let wire = Message::builder("SETNAME")
        .prefix(&mask)
        .trailing(realname)
        .to_wire();
    // `co_members` is the deduped, self-excluded set across every shared channel.
    for member in ctx.channels.co_members(uid) {
        if !ctx.is_local_uid(&member) {
            continue;
        }
        if ctx.registry.caps_of(&member).setname {
            ctx.registry.deliver_uid(&member, wire.clone());
        }
    }
    // IRCv3 extended-monitor: also send the SETNAME line to MONITOR watchers (extended-monitor +
    // setname) that share no channel with the changed user. Co-members are deduped inside
    // notify_watchers.
    crate::monitor::notify_watchers(ctx, nick, uid, |c| c.setname, &wire);
}

/// Inbound `:<uid> ENCAP * SETNAME :<realname>` — a peer announcing a remote user's realname
/// change. The changed user is the message **source** (prefix uid), as in [`super::umode`].
/// Updates that user's [`RemoteUser`](super::RemoteUser) mirror + its registry record, then
/// notifies local capable co-members. Does **not** relay — the caller
/// ([`super::forward::inbound_encap`]) onward-relays every `ENCAP` verbatim (split-horizon).
///
/// An unknown source, or a missing/empty realname, leaves all state untouched (the line is still
/// relayed by the caller, so the rest of the network converges).
pub(crate) fn apply_encap_setname(_link: &PeerLink, msg: &Message, ctx: &ServerContext) {
    let Some(uid) = msg.prefix().and_then(|p| Uid::try_from(p).ok()) else {
        return;
    };
    // The realname is the trailing param (after the `*` target-mask and `SETNAME` subcommand
    // verb); fall back to the first middle param after them for a single-word value.
    let realname = msg
        .trailing()
        .or_else(|| msg.params().get(2).map(String::as_str))
        .filter(|s| !s.is_empty());
    let Some(realname) = realname else {
        return;
    };
    let Some(mut u) = ctx.net.user(&uid) else {
        return;
    };
    let (nick, user, host) = (u.nick.clone(), u.user.clone(), u.host.clone());
    u.realname = realname.to_string();
    ctx.net.insert_user(u);
    ctx.registry.set_realname(&uid, realname);
    notify_co_members(ctx, &uid, &nick, &user, &host, realname);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cap::ClientCaps;
    use crate::ident::{ServerName, Sid};
    use crate::registry::Envelope;
    use crate::s2s::network::{RemoteServer, RemoteUser};
    use std::sync::Arc;
    use tokio::sync::mpsc::UnboundedReceiver;

    fn ctx() -> Arc<ServerContext> {
        let kdl = r#"
            server { name "leveva.test"; description "Leveva Test"; sid "0ABC" }
            admin { name "A"; email "a@test"; network "Net" }
            class "c" { ping-freq 90; max-links 10; sendq 1000 }
        "#;
        let cfg = crate::config::Config::from_kdl(kdl).expect("valid test config");
        ServerContext::from_config(&cfg, "now".to_string(), None)
    }

    fn link_peer(
        ctx: &ServerContext,
        sid_s: &str,
        name_s: &str,
    ) -> (PeerLink, UnboundedReceiver<Envelope>) {
        let (link, rx) = PeerLink::new(
            Sid::try_from(sid_s).unwrap(),
            ServerName::try_from(name_s).unwrap(),
        );
        ctx.net.insert_server(RemoteServer {
            sid: link.sid.clone(),
            name: link.name.clone(),
            description: "Peer".into(),
            hopcount: 1,
            uplink: link.sid.clone(),
            parent: ctx.uids.sid().clone(),
        });
        ctx.peers
            .link(link.sid.clone(), link.mailbox_tx.clone(), link.caps);
        (link, rx)
    }

    /// Introduce a remote user behind `link`; returns its UID.
    fn introduce(ctx: &ServerContext, link: &PeerLink, uid_s: &str, nick: &str) -> Uid {
        let uid = Uid::try_from(uid_s).unwrap();
        ctx.net.insert_user(RemoteUser {
            uid: uid.clone(),
            nick: nick.into(),
            user: "ruser".into(),
            host: "r.host".into(),
            ip: "0".into(),
            umodes: 0,
            realname: "old real".into(),
            server: link.sid.clone(),
        });
        ctx.registry
            .try_claim(
                &uid,
                nick,
                "ruser",
                "r.host",
                "old real",
                link.mailbox_tx.clone(),
            )
            .expect("fresh nick");
        uid
    }

    /// Claim a local `setname`-capable client and join it + `remote` to `chan`. Returns its mailbox.
    fn local_capable_co_member(
        ctx: &ServerContext,
        uid_s: &str,
        nick: &str,
        chan: &str,
        remote: &Uid,
    ) -> UnboundedReceiver<Envelope> {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let uid = Uid::try_from(uid_s).unwrap();
        ctx.registry
            .try_claim(&uid, nick, "u", "h", "r", tx)
            .expect("fresh nick");
        ctx.registry.set_caps(
            &uid,
            ClientCaps {
                setname: true,
                ..Default::default()
            },
        );
        ctx.channels.join(chan, &uid, 0);
        ctx.channels.join(chan, remote, 0);
        rx
    }

    fn parse(line: &str) -> Message {
        Message::parse(line).unwrap()
    }

    fn delivered(rx: &mut UnboundedReceiver<Envelope>) -> Message {
        match rx.try_recv().expect("a frame") {
            Envelope::Wire(b) => Message::parse(&String::from_utf8_lossy(&b)).unwrap(),
            other => panic!("expected wire, got {other:?}"),
        }
    }

    #[test]
    fn inbound_updates_mirror_and_registry_and_notifies_a_local_co_member() {
        let ctx = ctx();
        let (peer, _prx) = link_peer(&ctx, "0XYZ", "a.peer.test");
        let rover = introduce(&ctx, &peer, "0XYZAAAAA", "rover");
        let mut bob = local_capable_co_member(&ctx, "0ABCAAAAB", "bob", "#c", &rover);

        apply_encap_setname(
            &peer,
            &parse(":0XYZAAAAA ENCAP * SETNAME :New Real Name"),
            &ctx,
        );

        // Mirror + registry both reflect the new realname; identity is untouched.
        let u = ctx.net.user(&rover).unwrap();
        assert_eq!(u.realname, "New Real Name");
        assert_eq!((u.user.as_str(), u.host.as_str()), ("ruser", "r.host"));
        let rec = ctx.registry.record_of(&rover).unwrap();
        assert_eq!(rec.realname, "New Real Name");
        assert_eq!(rec.nick, "rover");
        // The local capable co-member sees the SETNAME with the current mask.
        let m = delivered(&mut bob);
        assert_eq!(m.command(), "SETNAME");
        assert_eq!(m.prefix(), Some("rover!ruser@r.host"));
        assert_eq!(m.trailing(), Some("New Real Name"));
    }

    #[test]
    fn inbound_for_an_unknown_source_changes_nothing() {
        let ctx = ctx();
        let (peer, _prx) = link_peer(&ctx, "0XYZ", "a.peer.test");
        let before = ctx.net.user_count();
        apply_encap_setname(&peer, &parse(":0XYZGHOST ENCAP * SETNAME :ghost"), &ctx);
        assert_eq!(ctx.net.user_count(), before, "no user fabricated");
    }

    #[test]
    fn inbound_with_missing_realname_is_a_noop() {
        let ctx = ctx();
        let (peer, _prx) = link_peer(&ctx, "0XYZ", "a.peer.test");
        let rover = introduce(&ctx, &peer, "0XYZAAAAA", "rover");
        apply_encap_setname(&peer, &parse(":0XYZAAAAA ENCAP * SETNAME"), &ctx);
        assert_eq!(ctx.net.user(&rover).unwrap().realname, "old real");
    }

    #[test]
    fn inbound_does_not_notify_a_non_cap_co_member() {
        let ctx = ctx();
        let (peer, _prx) = link_peer(&ctx, "0XYZ", "a.peer.test");
        let rover = introduce(&ctx, &peer, "0XYZAAAAA", "rover");
        let (tx, mut mel) = tokio::sync::mpsc::unbounded_channel();
        let mel_uid = Uid::try_from("0ABCAAAAB").unwrap();
        ctx.registry
            .try_claim(&mel_uid, "mel", "u", "h", "r", tx)
            .unwrap();
        ctx.channels.join("#c", &mel_uid, 0);
        ctx.channels.join("#c", &rover, 0);
        apply_encap_setname(
            &peer,
            &parse(":0XYZAAAAA ENCAP * SETNAME :New Real Name"),
            &ctx,
        );
        // The realname still updates, but a non-cap co-member gets nothing (no fallback).
        assert_eq!(ctx.net.user(&rover).unwrap().realname, "New Real Name");
        assert!(mel.try_recv().is_err(), "no SETNAME to a non-cap client");
    }

    #[test]
    fn local_setname_broadcasts_to_every_peer() {
        let ctx = ctx();
        let (_a, mut a_rx) = link_peer(&ctx, "0XYZ", "a.peer.test");
        let (_b, mut b_rx) = link_peer(&ctx, "0DEF", "b.peer.test");
        let uid = Uid::try_from("0ABCAAAAA").unwrap();
        local_setname(&ctx, &uid, "Bruce Wayne");
        let want = ":0ABCAAAAA ENCAP * SETNAME :Bruce Wayne\r\n";
        assert_eq!(delivered(&mut a_rx).to_wire(), want.as_bytes());
        assert_eq!(delivered(&mut b_rx).to_wire(), want.as_bytes());
    }

    #[test]
    fn local_setname_with_no_peers_is_a_noop() {
        let ctx = ctx();
        let uid = Uid::try_from("0ABCAAAAA").unwrap();
        local_setname(&ctx, &uid, "Bruce Wayne");
    }
}
//! **Services account assignment (`ENCAP SU`)** — the inbound `:<sid> ENCAP * SU <uid>
//! :<account>` carrier that logs a user in to (or out of) a services account (P11 slice 139).
//!
//! leveva keeps **no account database of its own**: a user acquires an account name only when a
//! remote server (services) broadcasts `ENCAP * SU` across the link (the user's explicit
//! directive). Unlike [`super::setname`]/[`super::umode`] — whose changed user is the message
//! *source* — `SU` targets another user, so the changed user is the **`<uid>` parameter** (a
//! service setting someone else's login), and the prefix is just the announcing server.
//!
//! [`apply_encap_su`] sets (or clears) the target's account on its registry record — the single
//! source of truth the read plane uses (WHOIS `330`, WHO/WHOX `%a`); UNICK has already mirrored
//! every remote user into the registry, so a local-or-remote target is handled uniformly. It
//! does **not** relay — the caller ([`super::forward::inbound_encap`]) onward-relays every
//! `ENCAP` verbatim (split-horizon), so the rest of the net converges.
//!
//! # Account string
//!
//! - **Present + valid** trailing → login: stored verbatim (services author the name).
//! - **Absent / empty** trailing → logout: the account is cleared.
//! - **Invalid** (would not survive as a wire middle param — contains SP/CR/LF or a leading
//!   `:`) → left untouched (the line still relays; a well-behaved service never sends one).
//!
//! # Divergences (documented)
//!
//! - **`account-notify` push (slice 140)** — a login/logout now pushes `:nick!user@host ACCOUNT
//!   <account|*>` to local capable co-members (see [`notify_account`]). The companion read-plane
//!   caps — the `@account` message tag (`account-tag`, slice 141) and login-on-join
//!   (`extended-join`, slice 142) — also read this services-account SoT, completing the IRCv3
//!   account-extensions trio. They live at their own delivery planes ([`crate::command::message`]/
//!   [`crate::command::tagmsg`] and [`crate::command::join`]).
//! - **No `RemoteUser` mirror field** — nothing reads it yet (the read plane uses the registry
//!   record); it is added in the burst-re-assertion follow-on when a netburst must carry account.
//! - **No `leveva-integration` differential** — the C 2.11 oracle's `ENCAP` is an opaque
//!   placeholder relay with no `SU` interpretation.

use crate::ident::Uid;
use crate::server::ServerContext;
use crate::Message;

use super::PeerLink;

/// Whether `account` is usable as an IRC middle parameter on the read plane (WHO `%a`) and the
/// WHOIS `330` middle field: non-empty, no SPACE/CR/LF, and no leading `:` (which would re-parse
/// as a trailing). A well-behaved service never sends an invalid name; this is purely defensive.
pub fn account_valid(account: &str) -> bool {
    !account.is_empty()
        && !account.starts_with(':')
        && !account
            .bytes()
            .any(|b| b == b' ' || b == b'\r' || b == b'\n')
}

/// Inbound `:<sid> ENCAP * SU <uid> [:<account>]` — a service logging `<uid>` in to (or out of)
/// an account. Sets/clears the account on the target's registry record. Does **not** relay (the
/// caller onward-relays every `ENCAP` verbatim). An unparseable/unknown target uid is a no-op
/// (the registry setter no-ops on an unknown uid).
pub(crate) fn apply_encap_su(_link: &PeerLink, msg: &Message, ctx: &ServerContext) {
    // Params after the ENCAP target-mask (`*`) and the `SU` verb: [`*`, `SU`, `<uid>`].
    let Some(uid) = msg
        .params()
        .get(2)
        .and_then(|p| Uid::try_from(p.as_str()).ok())
    else {
        return;
    };
    // The account is the trailing param, or a single-word middle param after the uid.
    let account = msg
        .trailing()
        .or_else(|| msg.params().get(3).map(String::as_str))
        .filter(|s| !s.is_empty());
    match account {
        // Login — but only with a wire-safe name; a malformed one leaves state untouched and
        // notifies nothing.
        Some(a) if account_valid(a) => {
            ctx.registry.set_account(&uid, Some(a));
            notify_account(ctx, &uid, Some(a));
        }
        Some(_) => {}
        // Logout — absent/empty trailing clears the account; the push carries `*`.
        None => {
            ctx.registry.set_account(&uid, None);
            notify_account(ctx, &uid, None);
        }
    }
}

/// IRCv3 **account-notify** push (P11 slice 140): tell every **local** channel co-member of
/// `uid` that negotiated `account-notify` that the user's services account changed, as a single
/// `:nick!user@host ACCOUNT <account>` — `<account>` being the name on login or the literal `*`
/// on logout. Also reaches `extended-monitor` watchers (gated `account_notify` + `extended_
/// monitor`) that share no channel. The changed user's mask is read from its registry record (the
/// account SoT, holding both local and remote users); an unknown user is a no-op (nothing visible
/// to notify). Mirrors [`super::away::notify_co_members`].
fn notify_account(ctx: &ServerContext, uid: &Uid, account: Option<&str>) {
    let Some(rec) = ctx.registry.record_of(uid) else {
        return;
    };
    let mask = format!("{}!{}@{}", rec.nick, rec.user, rec.host);
    let wire = Message::builder("ACCOUNT")
        .prefix(&mask)
        .param(account.unwrap_or("*"))
        .to_wire();
    for member in ctx.channels.co_members(uid) {
        if ctx.is_local_uid(&member) && ctx.registry.caps_of(&member).account_notify {
            ctx.registry.deliver_uid(&member, wire.clone());
        }
    }
    // Extended-monitor watchers (account-notify + extended-monitor) sharing no channel. The
    // changed user's co-members are deduped inside notify_watchers.
    crate::monitor::notify_watchers(ctx, &rec.nick, uid, |c| c.account_notify, &wire);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ident::{ServerName, Sid};
    use crate::registry::Envelope;
    use crate::s2s::network::RemoteServer;
    use std::sync::Arc;
    use tokio::sync::mpsc::UnboundedReceiver;

    fn ctx() -> Arc<ServerContext> {
        let kdl = r#"
            server { name "leveva.test"; description "Leveva Test"; sid "0ABC" }
            admin { name "A"; email "a@test"; network "Net" }
            class "c" { ping-freq 90; max-links 10; sendq 1000 }
        "#;
        let cfg = crate::config::Config::from_kdl(kdl).expect("valid test config");
        ServerContext::from_config(&cfg, "now".to_string(), None)
    }

    fn link_peer(ctx: &ServerContext) -> (PeerLink, UnboundedReceiver<Envelope>) {
        let (link, rx) = PeerLink::new(
            Sid::try_from("0XYZ").unwrap(),
            ServerName::try_from("a.peer.test").unwrap(),
        );
        ctx.net.insert_server(RemoteServer {
            sid: link.sid.clone(),
            name: link.name.clone(),
            description: "Peer".into(),
            hopcount: 1,
            uplink: link.sid.clone(),
            parent: ctx.uids.sid().clone(),
        });
        ctx.peers
            .link(link.sid.clone(), link.mailbox_tx.clone(), link.caps);
        (link, rx)
    }

    /// Claim a (remote-style) user in the registry; returns its UID.
    fn claim(ctx: &ServerContext, uid_s: &str, nick: &str) -> Uid {
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let uid = Uid::try_from(uid_s).unwrap();
        ctx.registry
            .try_claim(&uid, nick, "u", "h", "real", tx)
            .expect("fresh nick");
        uid
    }

    fn parse(line: &str) -> Message {
        Message::parse(line).unwrap()
    }

    #[test]
    fn su_logs_a_user_in() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let rover = claim(&ctx, "0XYZAAAAA", "rover");
        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :RoverAcct"), &ctx);
        assert_eq!(
            ctx.registry.record_of(&rover).unwrap().account.as_deref(),
            Some("RoverAcct")
        );
    }

    #[test]
    fn su_with_empty_trailing_logs_a_user_out() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let rover = claim(&ctx, "0XYZAAAAA", "rover");
        ctx.registry.set_account(&rover, Some("RoverAcct"));
        // Inverse: a logout (empty trailing) clears the previously-set account.
        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :"), &ctx);
        assert_eq!(ctx.registry.record_of(&rover).unwrap().account, None);
    }

    #[test]
    fn su_with_no_trailing_logs_a_user_out() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let rover = claim(&ctx, "0XYZAAAAA", "rover");
        ctx.registry.set_account(&rover, Some("RoverAcct"));
        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA"), &ctx);
        assert_eq!(ctx.registry.record_of(&rover).unwrap().account, None);
    }

    #[test]
    fn su_for_an_unknown_uid_is_a_noop() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        // No user claimed; must not panic and must register nothing.
        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :Acct"), &ctx);
        assert!(ctx
            .registry
            .record_of(&Uid::try_from("0XYZAAAAA").unwrap())
            .is_none());
    }

    #[test]
    fn su_with_an_invalid_account_leaves_state_untouched() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let rover = claim(&ctx, "0XYZAAAAA", "rover");
        ctx.registry.set_account(&rover, Some("Good"));
        // A space in the (middle-param) account can't be a wire param → ignored, prior kept.
        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :bad name"), &ctx);
        // The whole "bad name" is the trailing, which contains a space → invalid → no-op.
        assert_eq!(
            ctx.registry.record_of(&rover).unwrap().account.as_deref(),
            Some("Good")
        );
    }

    #[test]
    fn account_valid_predicate() {
        assert!(account_valid("Alice"));
        assert!(account_valid("a_b-c.123"));
        assert!(!account_valid(""));
        assert!(!account_valid(":lead"));
        assert!(!account_valid("two words"));
        assert!(!account_valid("cr\rlf"));
    }

    // ---- account-notify (slice 140): the ACCOUNT co-member push -------------------------------

    use crate::cap::ClientCaps;

    /// Claim a (remote-style) user and join it to `chan` so the account-notify fan-out has a mask
    /// + co-members to reach. Returns its UID.
    fn claim_in_chan(ctx: &ServerContext, uid_s: &str, nick: &str, chan: &str) -> Uid {
        let uid = claim(ctx, uid_s, nick);
        ctx.channels.join(chan, &uid, 0);
        uid
    }

    /// Claim a local client (optionally with `account-notify`), join it to `chan`, return its
    /// mailbox so the test can observe what it was delivered.
    fn local_member(
        ctx: &ServerContext,
        uid_s: &str,
        nick: &str,
        chan: &str,
        account_notify: bool,
    ) -> UnboundedReceiver<Envelope> {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let uid = Uid::try_from(uid_s).unwrap();
        ctx.registry
            .try_claim(&uid, nick, "u", "h", "r", tx)
            .expect("fresh nick");
        ctx.registry.set_caps(
            &uid,
            ClientCaps {
                account_notify,
                ..Default::default()
            },
        );
        ctx.channels.join(chan, &uid, 0);
        rx
    }

    fn delivered(rx: &mut UnboundedReceiver<Envelope>) -> Message {
        match rx.try_recv().expect("a frame") {
            Envelope::Wire(b) => Message::parse(&String::from_utf8_lossy(&b)).unwrap(),
            other => panic!("expected wire, got {other:?}"),
        }
    }

    #[test]
    fn su_login_notifies_a_local_capable_co_member() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let rover = claim_in_chan(&ctx, "0XYZAAAAA", "rover", "#c");
        let mut carol = local_member(&ctx, "0ABCAAAAB", "carol", "#c", true);

        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :RoverAcct"), &ctx);

        assert_eq!(
            ctx.registry.record_of(&rover).unwrap().account.as_deref(),
            Some("RoverAcct")
        );
        let m = delivered(&mut carol);
        assert_eq!(m.command(), "ACCOUNT");
        assert_eq!(m.prefix(), Some("rover!u@h"));
        assert_eq!(m.params().first().map(String::as_str), Some("RoverAcct"));
    }

    #[test]
    fn su_logout_notifies_with_a_star() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let rover = claim_in_chan(&ctx, "0XYZAAAAA", "rover", "#c");
        let mut carol = local_member(&ctx, "0ABCAAAAB", "carol", "#c", true);

        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :RoverAcct"), &ctx);
        let _ = delivered(&mut carol); // the login push

        // Inverse: a logout (empty trailing) pushes `ACCOUNT *`.
        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :"), &ctx);
        assert_eq!(ctx.registry.record_of(&rover).unwrap().account, None);
        let m = delivered(&mut carol);
        assert_eq!(m.command(), "ACCOUNT");
        assert_eq!(
            m.params().first().map(String::as_str),
            Some("*"),
            "logout → *"
        );
    }

    #[test]
    fn su_does_not_notify_a_non_cap_co_member() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let _rover = claim_in_chan(&ctx, "0XYZAAAAA", "rover", "#c");
        let mut mel = local_member(&ctx, "0ABCAAAAB", "mel", "#c", false);

        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :RoverAcct"), &ctx);
        assert!(
            mel.try_recv().is_err(),
            "a non-cap co-member gets no ACCOUNT"
        );
    }

    #[test]
    fn su_for_an_unknown_uid_notifies_nothing() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let mut carol = local_member(&ctx, "0ABCAAAAB", "carol", "#c", true);
        // No rover claimed → no record, no mask, no push.
        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :Acct"), &ctx);
        assert!(
            carol.try_recv().is_err(),
            "unknown target → nothing delivered"
        );
    }

    #[test]
    fn su_with_an_invalid_account_does_not_notify() {
        let ctx = ctx();
        let (peer, _rx) = link_peer(&ctx);
        let rover = claim_in_chan(&ctx, "0XYZAAAAA", "rover", "#c");
        ctx.registry.set_account(&rover, Some("Good"));
        let mut carol = local_member(&ctx, "0ABCAAAAB", "carol", "#c", true);

        apply_encap_su(&peer, &parse(":0XYZ ENCAP * SU 0XYZAAAAA :bad name"), &ctx);
        // State untouched and nothing pushed (a space-bearing account is invalid).
        assert_eq!(
            ctx.registry.record_of(&rover).unwrap().account.as_deref(),
            Some("Good")
        );
        assert!(carol.try_recv().is_err(), "invalid account → no push");
    }
}
==== forward.rs inbound_encap ====
//! **Server-sourced relay & notice verbs** — `ENCAP`, `SDIE`, `ERROR`.
//!
//! Three of the `msgtab` rows whose client column is `m_nop` (server-sourced only): a linked
//! peer sends them, and leveva either **forwards** the line across the network or **absorbs**
//! it. None mutates network topology or user state, so they sit apart from the heavier
//! `SMASK` (masked-server introduction) and `UMODE` (remote user-mode propagation) S2S rows.
//!
//! | verb    | behaviour                                                                       |
//! |---------|---------------------------------------------------------------------------------|
//! | `ENCAP` | opaque encapsulated command — onward-relay **verbatim**, split-horizon          |
//! | `SDIE`  | placeholder — onward-relay verbatim, split-horizon; **does not** terminate      |
//! | `ERROR` | inbound peer error — surface via `tracing::warn!`, **no relay**, no reply       |
//!
//! # `SDIE` is a placeholder, not a kill
//!
//! `ENCAP`/`SDIE` are placeholders in IRCnet 2.11 (the upstream commit that introduced them
//! is titled *"new protocol commands ENCAP and SDIE (just placeholders)"*); the oracle
//! `m_sdie` body is only `sendto_serv_v(...); return 0;` — no `s_die()`. leveva mirrors that:
//! `SDIE` is a pure relay. Routing it to the [`control`](crate::control) plane would be a
//! divergence from parity, not parity itself (and `golden_s2s_forward` guards the regression).
//!
//! # Divergences from the oracle (documented)
//!
//! - **Verbatim forward vs. reconstruction.** The oracle rebuilds the `ENCAP` buffer
//!   field-by-field (its `i>=3 && i==parc-1` trailing-colon rule). leveva forwards
//!   [`Message::to_wire`] — the originating server already framed the line and [`Message`]
//!   round-trips prefix/params/trailing faithfully, the established [`super::relay`] onward
//!   convention. The trailing-colon normalisation thus follows leveva's parser; the payload
//!   is opaque, so there is no semantic difference.
//! - **No `SCH_ERROR` notice channel.** leveva has no server-notice channel, so the oracle's
//!   `sendto_flag(SCH_ERROR, …)` becomes a `tracing::warn!` for `ERROR`, and the (unreachable,
//!   512-capped) `ENCAP too long` notice is dropped.
//! - **Single-uplink no-op.** With one direct peer, split-horizon forwarding reaches zero
//!   peers — the same documented single-server divergence as all of [`super::relay`].
//! - **No `leveva-integration` differential** — the oracle handlers are impure
//!   (`sendto_serv_v`/`sendto_flag` into global buffers), so there is no pure entry point.

use crate::server::ServerContext;
use crate::Message;

use super::PeerLink;

/// Inbound `:<sid> ENCAP <args…>` — an opaque encapsulated command. leveva does not interpret
/// the encapsulated subcommand (the oracle is itself a placeholder relay); it onward-relays
/// the line verbatim to every peer **except** the one it arrived on (split-horizon).
pub fn inbound_encap(link: &mut PeerLink, msg: &Message, ctx: &ServerContext) -> Vec<u8> {
    // CAPAB is **link-local**: it announces the *direct* peer's capabilities (chiefly whether
    // it accepts message tags on S2S lines), so it is interpreted and **not** relayed onward —
    // capabilities describe the transport hop, not the wider net. Record it on both the live
    // [`PeerLink`] and the [`PeerLinks`](super::links::PeerLinks) routing table (the authority
    // the relay tag-gating reads). See [`super::caps`].
    if msg.params().get(1).map(String::as_str) == Some("CAPAB") {
        let tokens = msg
            .trailing()
            .or_else(|| msg.params().get(2).map(String::as_str))
            .unwrap_or("");
        let caps = super::caps::PeerCaps::from_capab(tokens);
        link.caps = caps;
        ctx.peers.set_caps(&link.sid, caps);
        return Vec::new();
    }
    // USERTS (P11 slice 133) carries the timestamp of the user introduced by the *very next*
    // `UNICK` (`:<uid> ENCAP * USERTS <ts>`). Buffer it on the link keyed by UID, so
    // `handle_unick` has the newcomer's timestamp the moment its `UNICK` arrives. Link-local
    // like CAPAB — a per-user hint consumed locally, not relayed onward.
    if msg.params().get(1).map(String::as_str) == Some("USERTS") {
        super::unick::buffer_user_ts(link, msg);
        return Vec::new();
    }
    // SASL relay (P11 slice 159): a `SASL` continuation/verdict or a `SVSLOGIN` from the services
    // agent is addressed to *us* — consume it and route the result back to the in-flight client;
    // do **not** onward-relay (unlike the interpreted subcommands below).
    match msg.params().get(1).map(String::as_str) {
        Some("SASL") => {
            super::sasl::apply_encap_sasl(link, msg, ctx);
            return Vec::new();
        }
        Some("SVSLOGIN") => {
            super::sasl::apply_encap_svslogin(link, msg, ctx);
            return Vec::new();
        }
        _ => {}
    }
    // A couple of encapsulated subcommands are interpreted; everything else stays an opaque
    // placeholder. Either way the line is onward-relayed verbatim (split-horizon), so the
    // rest of the network converges.
    match msg.params().get(1).map(String::as_str) {
        // The IRCv3 host-change relay (P11 slice 71): mutate the changed user's mirror +
        // notify local capable co-members.
        Some("CHGHOST") => super::chghost::apply_encap_chghost(link, msg, ctx),
        // The CertFP relay (P11 slice 78): record the source user's certificate fingerprint
        // so a cross-server WHOIS shows its `276`.
        Some("CERTFP") => super::certfp::apply_encap_certfp(link, msg, ctx),
        // The IRCv3 realname-change relay (P11 slice 90): mutate the changed user's mirror +
        // notify local capable co-members.
        Some("SETNAME") => super::setname::apply_encap_setname(link, msg, ctx),
        // The IRCv3 message-redaction relay (P11 slice 127, reworked from the slice-125
        // first-class verb): deliver the client-form REDACT to local capable members/nick.
        Some("REDACT") => super::redact::apply_encap_redact(link, msg, ctx),
        // The channel-timestamp merge arbitration (P11 slice 132, leveva-native): reconcile a
        // channel's creation TS (oldest wins) and relay the resulting deop to local members.
        Some("CHANTS") => super::chants::apply_encap_chants(link, msg, ctx),
        // Services account assignment (P11 slice 139): log the target `<uid>` in to / out of an
        // account so a cross-server WHOIS shows its `330` and the WHO/WHOX `%a` field.
        Some("SU") => super::su::apply_encap_su(link, msg, ctx),
        _ => {}
    }
    ctx.peers.broadcast(Some(&link.sid), msg.to_wire());
    Vec::new()
}

/// Inbound `:<sid> SDIE` — the placeholder "silent die" announcement. A pure relay: forward
/// verbatim to every other peer (split-horizon). It does **not** terminate this server.
pub fn inbound_sdie(link: &mut PeerLink, msg: &Message, ctx: &ServerContext) -> Vec<u8> {
    ctx.peers.broadcast(Some(&link.sid), msg.to_wire());
    Vec::new()
}

/// Inbound `:<sid> ERROR :<text>` — an error report from a linked peer. The oracle reports it
/// to the `SCH_ERROR` server-notice channel and returns without relaying; leveva has no such
