citadel_sdk/

remote_ext.rs

//! Remote Protocol Extensions
//!
//! This module extends the core NodeRemote functionality with high-level operations
//! for managing connections, file transfers, and peer interactions in the Citadel
//! Protocol network.
//!
//! # Features
//! - User registration and authentication
//! - Connection management
//! - File transfer operations
//! - Virtual filesystem support
//! - Peer discovery and management
//! - Group communication
//! - Security settings configuration
//!
//! # Example
//! ```rust
//! use citadel_sdk::prelude::*;
//!
//! async fn example<R: Ratchet>(remote: NodeRemote<R>) -> Result<(), NetworkError> {
//!     // Register a new user
//!     let reg = remote.register_with_defaults(
//!         "127.0.0.1:25021",
//!         "John Doe",
//!         "john.doe",
//!         "password123"
//!     ).await?;
//!
//!     // Connect to a peer
//!     let auth = AuthenticationRequest::credentialed("john.doe", "password123");
//!     let conn = remote.connect_with_defaults(auth).await?;
//!
//!     // Send a file to a peer
//!     remote.find_target("john.doe", "peer.name")
//!         .await?
//!         .send_file("/path/to/file.txt")
//!         .await?;
//!
//!     Ok(())
//! }
//! ```
//!
//! # Important Notes
//! - All operations are asynchronous
//! - Connections are automatically managed
//! - File transfers support chunking
//! - Virtual filesystem is encrypted
//! - Peer connections require mutual registration
//!
//! # Related Components
//! - [`NodeRemote`]: Core remote interface
//! - [`ClientServerRemote`]: Client-server communication
//! - [`PeerRemote`]: Peer-to-peer communication
//! - [`CitadelClientServerConnection`]: Connection management
//! - [`RegisterSuccess`]: Registration handling
//!
//! [`NodeRemote`]: crate::prelude::NodeRemote
//! [`ClientServerRemote`]: crate::prelude::ClientServerRemote
//! [`PeerRemote`]: crate::prelude::PeerRemote
//! [`CitadelClientServerConnection`]: crate::prelude::CitadelClientServerConnection
//! [`RegisterSuccess`]: crate::prelude::RegisterSuccess

use crate::prefabs::ClientServerRemote;
use crate::prelude::results::{PeerConnectSuccess, PeerRegisterStatus};
use crate::prelude::*;
use crate::remote_ext::remote_specialization::PeerRemote;
use crate::remote_ext::results::LocalGroupPeerFullInfo;
use std::ops::{Deref, DerefMut};

use futures::StreamExt;
use std::path::PathBuf;
use std::time::Duration;

pub(crate) mod user_ids {
    use crate::prelude::*;
    use std::ops::Deref;

    #[derive(Debug)]
    /// A reference to a user identifier
    pub struct SymmetricIdentifierHandleRef<'a, R: Ratchet> {
        pub(crate) user: VirtualTargetType,
        pub(crate) remote: &'a NodeRemote<R>,
        pub(crate) target_username: Option<String>,
    }

    impl<R: Ratchet> SymmetricIdentifierHandleRef<'_, R> {
        pub fn into_owned(self) -> SymmetricIdentifierHandle<R> {
            SymmetricIdentifierHandle {
                user: self.user,
                remote: self.remote.clone(),
                target_username: self.target_username,
            }
        }
    }

    #[derive(Clone, Debug)]
    /// A convenience structure for executing commands that depend on a specific registered user
    pub struct SymmetricIdentifierHandle<R: Ratchet> {
        user: VirtualTargetType,
        remote: NodeRemote<R>,
        target_username: Option<String>,
    }

    pub trait TargetLockedRemote<R: Ratchet>: Send + Sync {
        fn user(&self) -> &VirtualTargetType;
        fn remote(&self) -> &NodeRemote<R>;
        fn target_username(&self) -> Option<&str>;
        fn user_mut(&mut self) -> &mut VirtualTargetType;
        fn session_security_settings(&self) -> Option<&SessionSecuritySettings>;
    }

    impl<R: Ratchet> TargetLockedRemote<R> for SymmetricIdentifierHandleRef<'_, R> {
        fn user(&self) -> &VirtualTargetType {
            &self.user
        }
        fn remote(&self) -> &NodeRemote<R> {
            self.remote
        }
        fn target_username(&self) -> Option<&str> {
            self.target_username.as_deref()
        }
        fn user_mut(&mut self) -> &mut VirtualTargetType {
            &mut self.user
        }

        fn session_security_settings(&self) -> Option<&SessionSecuritySettings> {
            None
        }
    }

    impl<R: Ratchet> TargetLockedRemote<R> for SymmetricIdentifierHandle<R> {
        fn user(&self) -> &VirtualTargetType {
            &self.user
        }
        fn remote(&self) -> &NodeRemote<R> {
            &self.remote
        }
        fn target_username(&self) -> Option<&str> {
            self.target_username.as_deref()
        }
        fn user_mut(&mut self) -> &mut VirtualTargetType {
            &mut self.user
        }

        fn session_security_settings(&self) -> Option<&SessionSecuritySettings> {
            None
        }
    }

    impl<R: Ratchet> From<SymmetricIdentifierHandleRef<'_, R>> for SymmetricIdentifierHandle<R> {
        fn from(this: SymmetricIdentifierHandleRef<'_, R>) -> Self {
            this.into_owned()
        }
    }

    impl<R: Ratchet> Deref for SymmetricIdentifierHandle<R> {
        type Target = NodeRemote<R>;

        fn deref(&self) -> &Self::Target {
            &self.remote
        }
    }

    impl<R: Ratchet> Deref for SymmetricIdentifierHandleRef<'_, R> {
        type Target = NodeRemote<R>;

        fn deref(&self) -> &Self::Target {
            self.remote
        }
    }
}

/// Contains the elements required to communicate with the adjacent node
pub struct CitadelClientServerConnection<R: Ratchet> {
    /// An interface to send ordered, reliable, and encrypted messages
    pub(crate) channel: Option<PeerChannel<R>>,
    pub remote: ClientServerRemote<R>,
    /// Only available if UdpMode was enabled at the beginning of a session
    pub udp_channel_rx: Option<citadel_io::tokio::sync::oneshot::Receiver<UdpChannel<R>>>,
    /// Contains the Google auth minted at the central server (if the central server enabled it), as well as any other services enabled by the central server
    pub services: ServicesObject,
    pub cid: u64,
    pub session_security_settings: SessionSecuritySettings,
    /// The server's path policy: which paths this session's peer connections may use, in which
    /// order, for which traffic. `None` from a server that predates sending one
    /// ([`PATH_POLICY_SINCE`](citadel_proto::constants::PATH_POLICY_SINCE)).
    pub path_policy: Option<PathPolicy>,
}

impl<R: Ratchet> CitadelClientServerConnection<R> {
    /// Splits the channel into a send and receive half. This will render
    /// the other fields of this connection object innaccessible
    ///
    /// # Panics
    ///  - If the channel has already been taken
    pub fn split(self) -> (PeerChannelSendHalf<R>, PeerChannelRecvHalf<R>) {
        self.channel.expect("Channel already taken").split()
    }

    pub fn take_channel(&mut self) -> Option<PeerChannel<R>> {
        self.channel.take()
    }
}

impl<R: Ratchet> Deref for CitadelClientServerConnection<R> {
    type Target = ClientServerRemote<R>;

    fn deref(&self) -> &Self::Target {
        &self.remote
    }
}

impl<R: Ratchet> DerefMut for CitadelClientServerConnection<R> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.remote
    }
}

/// Contains the elements entailed by a successful registration
pub struct RegisterSuccess {
    pub cid: u64,
    /// The recovery codes of a post-quantum registration, to show the user once (see
    /// `RegisterOkay::recovery_codes`). Empty for a legacy registration.
    pub recovery_codes: Vec<String>,
}

/// Waits for the outcome of a registration request.
async fn await_registration<R: Ratchet, Rem: Remote<R>>(
    remote: &Rem,
    register_request: NodeRequest,
) -> Result<RegisterSuccess, NetworkError> {
    let mut subscription = remote.send_callback_subscription(register_request).await?;
    while let Some(status) = subscription.next().await {
        match status.into_result()? {
            NodeResult::RegisterOkay(RegisterOkay {
                cid,
                recovery_codes,
                ..
            }) => {
                return Ok(RegisterSuccess {
                    cid,
                    recovery_codes,
                });
            }
            NodeResult::RegisterFailure(err) => {
                return Err(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteRegisterFailure,
                    err.error_message
                ));
            }
            NodeResult::Disconnect(err) => {
                return Err(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteDisconnected,
                    err.message
                ));
            }
            evt => {
                log::warn!(target: "citadel", "Invalid NodeResult for Register request received: {evt:?}");
            }
        }
    }

    Err(citadel_io::error!(
        citadel_io::ErrorCode::RemoteKernelStreamDied,
        "register"
    ))
}

/// The legacy credentials (for a server below 0.12) and the password a post-quantum registration
/// uses, trimmed the same way the legacy credentials trim it.
async fn registration_credentials<
    P: Into<String> + Send,
    V: Into<String> + Send,
    K: Into<SecBuffer>,
>(
    full_name: P,
    username: V,
    proposed_password: K,
) -> Result<(ProposedCredentials, SecBuffer), NetworkError> {
    let password: SecBuffer = proposed_password.into();
    let trimmed = ProposedCredentials::registration_password(&password);
    let creds = ProposedCredentials::new_register(full_name, username, password).await?;
    Ok((creds, trimmed))
}

#[async_trait]
/// Endows the [NodeRemote] with additional functions
pub trait ProtocolRemoteExt<R: Ratchet>: Remote<R> {
    /// Registers with custom settings
    /// Returns a ticket which is used to uniquely identify the request in the protocol
    async fn register<
        T: std::net::ToSocketAddrs + Send,
        P: Into<String> + Send,
        V: Into<String> + Send,
        K: Into<SecBuffer> + Send,
    >(
        &self,
        addr: T,
        full_name: P,
        username: V,
        proposed_password: K,
        default_security_settings: SessionSecuritySettings,
        server_password: Option<PreSharedKey>,
    ) -> Result<RegisterSuccess, NetworkError> {
        self.register_admitted(
            addr,
            full_name,
            username,
            proposed_password,
            default_security_settings,
            server_password,
            None,
        )
        .await
    }

    /// As [`register`](Self::register), carrying an admission token (a Turnstile response) for a
    /// server that asks fresh registrations for one. A refusal is a `PqSignInAdmissionRequired`
    /// or `PqSignInAdmissionFailed` error.
    #[allow(clippy::too_many_arguments)]
    async fn register_admitted<
        T: std::net::ToSocketAddrs + Send,
        P: Into<String> + Send,
        V: Into<String> + Send,
        K: Into<SecBuffer> + Send,
    >(
        &self,
        addr: T,
        full_name: P,
        username: V,
        proposed_password: K,
        default_security_settings: SessionSecuritySettings,
        server_password: Option<PreSharedKey>,
        admission: Option<String>,
    ) -> Result<RegisterSuccess, NetworkError> {
        let (creds, password) =
            registration_credentials(full_name, username, proposed_password).await?;
        let register_request = NodeRequest::RegisterToHypernode(RegisterToHypernode {
            remote_addr: addr.to_socket_addrs()?.next().ok_or(citadel_io::error!(
                citadel_io::ErrorCode::RemoteInvalidSocketAddr
            ))?,
            proposed_credentials: creds,
            static_security_settings: default_security_settings,
            session_password: server_password.unwrap_or_default(),
            endpoint: None,
            password: Some(password),
            admission,
        });

        await_registration(self, register_request).await
    }

    /// Registers with a server reached by a WebSocket URL (`wss://org.example.net/`, or
    /// `ws://host:port/path`) rather than at an address of its own: a server behind an HTTP edge,
    /// which is told apart from the edge's other servers by hostname and path. The account
    /// remembers the URL, so a later [`connect`](Self::connect) with credentials dials it too.
    #[cfg(not(target_family = "wasm"))]
    async fn register_to_endpoint<
        P: Into<String> + Send,
        V: Into<String> + Send,
        K: Into<SecBuffer> + Send,
    >(
        &self,
        endpoint: citadel_io::WebSocketEndpoint,
        full_name: P,
        username: V,
        proposed_password: K,
        default_security_settings: SessionSecuritySettings,
        server_password: Option<PreSharedKey>,
    ) -> Result<RegisterSuccess, NetworkError> {
        self.register_to_endpoint_admitted(
            endpoint,
            full_name,
            username,
            proposed_password,
            default_security_settings,
            server_password,
            None,
        )
        .await
    }

    /// As [`register_to_endpoint`](Self::register_to_endpoint), carrying an admission token.
    #[cfg(not(target_family = "wasm"))]
    #[allow(clippy::too_many_arguments)]
    async fn register_to_endpoint_admitted<
        P: Into<String> + Send,
        V: Into<String> + Send,
        K: Into<SecBuffer> + Send,
    >(
        &self,
        endpoint: citadel_io::WebSocketEndpoint,
        full_name: P,
        username: V,
        proposed_password: K,
        default_security_settings: SessionSecuritySettings,
        server_password: Option<PreSharedKey>,
        admission: Option<String>,
    ) -> Result<RegisterSuccess, NetworkError> {
        let remote_addr = endpoint.resolve().await?;
        let (creds, password) =
            registration_credentials(full_name, username, proposed_password).await?;
        let register_request = NodeRequest::RegisterToHypernode(RegisterToHypernode {
            remote_addr,
            proposed_credentials: creds,
            static_security_settings: default_security_settings,
            session_password: server_password.unwrap_or_default(),
            endpoint: Some(endpoint),
            password: Some(password),
            admission,
        });

        await_registration(self, register_request).await
    }

    /// The WebSocket URL the local account `cid` registered to (see
    /// [`register_to_endpoint`](Self::register_to_endpoint)), or `None` when its server is the
    /// address in its connection info. Behind an HTTP edge that address is one of the edge's,
    /// shared with every other server there, so it does not identify the server; this does.
    async fn server_endpoint(
        &self,
        cid: u64,
    ) -> Result<Option<citadel_io::WebSocketEndpoint>, NetworkError> {
        citadel_proto::prelude::load_server_endpoint(self.account_manager(), cid).await
    }

    /// Registers using the default settings. The default uses No Google FCM keys and the default session security settings
    /// Returns a ticket which is used to uniquely identify the request in the protocol
    async fn register_with_defaults<
        T: std::net::ToSocketAddrs + Send,
        P: Into<String> + Send,
        V: Into<String> + Send,
        K: Into<SecBuffer> + Send,
    >(
        &self,
        addr: T,
        full_name: P,
        username: V,
        proposed_password: K,
    ) -> Result<RegisterSuccess, NetworkError> {
        self.register(
            addr,
            full_name,
            username,
            proposed_password,
            Default::default(),
            Default::default(),
        )
        .await
    }

    /// Connects with custom settings
    /// Returns a ticket which is used to uniquely identify the request in the protocol
    async fn connect(
        &self,
        auth: AuthenticationRequest,
        connect_mode: ConnectMode,
        udp_mode: UdpMode,
        keep_alive_timeout: Option<Duration>,
        session_security_settings: SessionSecuritySettings,
        server_password: Option<PreSharedKey>,
    ) -> Result<CitadelClientServerConnection<R>, NetworkError> {
        let connect_request = NodeRequest::ConnectToHypernode(ConnectToHypernode {
            auth_request: auth,
            connect_mode,
            udp_mode,
            keep_alive_timeout: keep_alive_timeout.map(|r| r.as_secs()),
            session_security_settings,
            session_password: server_password.unwrap_or_default(),
        });

        let mut subscription = self.send_callback_subscription(connect_request).await?;
        let status = subscription.next().await.ok_or(citadel_io::error!(
            citadel_io::ErrorCode::RemoteKernelStreamDied,
            "connect"
        ))?;

        return match status.into_result()? {
            NodeResult::ConnectSuccess(ConnectSuccess {
                ticket: _,
                session_cid: cid,
                remote_addr: _,
                is_personal: _,
                v_conn_type,
                services,
                welcome_message: _,
                channel,
                udp_rx_opt: udp_channel_rx,
                session_security_settings,
                path_policy,
            }) => Ok(CitadelClientServerConnection {
                remote: ClientServerRemote::new(
                    v_conn_type,
                    self.remote_ref().clone(),
                    session_security_settings,
                    None,
                    None,
                ),
                channel: Some(*channel),
                udp_channel_rx,
                services,
                cid,
                session_security_settings,
                path_policy,
            }),
            NodeResult::ConnectFail(ConnectFail {
                error_message: err, ..
            }) => Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteConnectFailed,
                err
            )),
            NodeResult::Disconnect(err) => {
                return Err(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteDisconnected,
                    err.message
                ));
            }
            res => Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteConnectUnexpectedResponse,
                citadel_io::Dbg(res)
            )),
        };
    }

    /// Connects with the default settings
    /// If FCM keys were created during the registration phase, then those keys will be used for the session. If new FCM keys need to be used, consider using [`Self::connect`]
    ///
    /// Never displaces a session the server already holds for the account
    /// (`ConnectMode::Standard { force_login: false }`). A client reconnecting after its link
    /// dropped should use [`Self::connect`] with `force_login: true`: see [`ConnectMode`].
    async fn connect_with_defaults(
        &self,
        auth: AuthenticationRequest,
    ) -> Result<CitadelClientServerConnection<R>, NetworkError> {
        self.connect(
            auth,
            ConnectMode::Standard { force_login: false },
            Default::default(),
            None,
            Default::default(),
            Default::default(),
        )
        .await
    }

    /// Creates a valid target identifier used to make protocol requests. Raw user IDs or usernames can be used
    /// ```
    /// use citadel_sdk::prelude::*;
    /// # use citadel_sdk::prefabs::client::single_connection::SingleClientServerConnectionKernel;
    ///
    /// let server_connection_settings = DefaultServerConnectionSettingsBuilder::credentialed_login("127.0.0.1:25021", "john.doe", "password").build().unwrap();
    ///
    /// # SingleClientServerConnectionKernel::new(server_connection_settings, |conn| async move {
    /// conn.find_target("my_account", "my_peer").await?.send_file("/path/to/file.pdf").await
    /// // or: conn.find_target(1234, "my_peer").await? [...]
    /// # });
    /// ```
    async fn find_target<T: Into<UserIdentifier> + Send, P: Into<UserIdentifier> + Send>(
        &self,
        local_user: T,
        peer: P,
    ) -> Result<SymmetricIdentifierHandleRef<'_, R>, NetworkError> {
        let account_manager = self.account_manager();
        account_manager
            .find_target_information(local_user, peer)
            .await?
            .map(move |(cid, peer)| {
                if peer.parent_icid != 0 {
                    SymmetricIdentifierHandleRef {
                        user: VirtualTargetType::ExternalGroupPeer {
                            session_cid: cid,
                            interserver_cid: peer.parent_icid,
                            peer_cid: peer.cid,
                        },
                        remote: self.remote_ref(),
                        target_username: None,
                    }
                } else {
                    SymmetricIdentifierHandleRef {
                        user: VirtualTargetType::LocalGroupPeer {
                            session_cid: cid,
                            peer_cid: peer.cid,
                        },
                        remote: self.remote_ref(),
                        target_username: None,
                    }
                }
            })
            .ok_or_else(|| citadel_io::error!(citadel_io::ErrorCode::RemoteTargetPairNotFound))
    }

    /// Creates a proposed target from the valid local user to an unregistered peer in the network. Used when creating registration requests for peers.
    /// Currently only supports LocalGroup <-> LocalGroup peer connections
    async fn propose_target<T: Into<UserIdentifier> + Send, P: Into<UserIdentifier> + Send>(
        &self,
        local_user: T,
        peer: P,
    ) -> Result<SymmetricIdentifierHandleRef<'_, R>, NetworkError> {
        let local_cid = self.get_session_cid(local_user).await?;
        match peer.into() {
            UserIdentifier::ID(peer_cid) => Ok(SymmetricIdentifierHandleRef {
                user: VirtualTargetType::LocalGroupPeer {
                    session_cid: local_cid,
                    peer_cid,
                },
                remote: self.remote_ref(),
                target_username: None,
            }),
            UserIdentifier::Username(uname) => {
                let peer_cid = self
                    .remote_ref()
                    .account_manager()
                    .find_target_information(local_cid, uname.clone())
                    .await?
                    .map(|r| r.1.cid)
                    .unwrap_or(0);
                Ok(SymmetricIdentifierHandleRef {
                    user: VirtualTargetType::LocalGroupPeer {
                        session_cid: local_cid,
                        peer_cid,
                    },
                    remote: self.remote_ref(),
                    target_username: Some(uname),
                })
            }
        }
    }

    /// Returns a list of local group peers on the network for local_user. May or may not be registered to the user. To get a list of registered users to local_user, run [`Self::get_local_group_mutual_peers`]
    /// - limit: if None, all peers are obtained. If Some, at most the specified number of peers will be obtained
    async fn get_local_group_peers<T: Into<UserIdentifier> + Send>(
        &self,
        local_user: T,
        limit: Option<usize>,
    ) -> Result<Vec<LocalGroupPeerFullInfo>, NetworkError> {
        let local_cid = self.get_session_cid(local_user).await?;
        let command = NodeRequest::PeerCommand(PeerCommand {
            session_cid: local_cid,
            command: PeerSignal::GetRegisteredPeers {
                peer_conn_type: ClientConnectionType::Server {
                    session_cid: local_cid,
                },
                response: None,
                limit: limit.map(|r| r as i32),
            },
        });

        let mut stream = self.send_callback_subscription(command).await?;

        while let Some(status) = stream.next().await {
            if let NodeResult::PeerEvent(PeerEvent {
                event:
                    PeerSignal::GetRegisteredPeers {
                        peer_conn_type: _,
                        response: Some(PeerResponse::RegisteredCids(peer_info, is_onlines)),
                        limit: _,
                    },
                ..
            }) = status.into_result()?
            {
                return Ok(peer_info
                    .into_iter()
                    .zip(is_onlines)
                    .filter_map(|(peer_info, is_online)| {
                        peer_info.map(|info| LocalGroupPeerFullInfo {
                            cid: info.cid,
                            username: Some(info.username),
                            full_name: Some(info.full_name),
                            is_online,
                        })
                    })
                    .collect());
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteKernelStreamDied,
            "get_local_group_peers"
        ))
    }

    /// Returns a list of mutually-registered peers with the local_user
    async fn get_local_group_mutual_peers<T: Into<UserIdentifier> + Send>(
        &self,
        local_user: T,
    ) -> Result<Vec<LocalGroupPeerFullInfo>, NetworkError> {
        let local_cid = self.get_session_cid(local_user).await?;
        let command = NodeRequest::PeerCommand(PeerCommand {
            session_cid: local_cid,
            command: PeerSignal::GetMutuals {
                v_conn_type: ClientConnectionType::Server {
                    session_cid: local_cid,
                },
                response: None,
            },
        });

        let mut stream = self.send_callback_subscription(command).await?;

        while let Some(status) = stream.next().await {
            if let NodeResult::PeerEvent(PeerEvent {
                event:
                    PeerSignal::GetMutuals {
                        v_conn_type: _,
                        response: Some(PeerResponse::RegisteredCids(peer_info, is_onlines)),
                    },
                ..
            }) = status.into_result()?
            {
                return Ok(peer_info
                    .into_iter()
                    .zip(is_onlines)
                    .filter_map(|(peer_info, is_online)| {
                        peer_info.map(|info| LocalGroupPeerFullInfo {
                            cid: info.cid,
                            username: Some(info.username),
                            full_name: Some(info.full_name),
                            is_online,
                        })
                    })
                    .collect());
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteSessionStreamDied
        ))
    }

    /// Round-trips an authenticated probe to the server of session `cid` now, rather than on the
    /// keep-alive schedule (15 minutes, 45 to time out): `Ok(rtt)`, `Timeout` when no answer came
    /// within `timeout`, or `Error` when it could not be sent (no such connected session, or a
    /// server below protocol 0.12.1, which cannot answer one). Over a WebSocket the probe also
    /// sends a WebSocket ping.
    async fn probe_server(&self, cid: u64, timeout: Duration) -> ServerProbeOutcome {
        crate::server_probe::probe_server(self, cid, timeout).await
    }

    /// Ends session `cid` here and now, without asking the server: its transport is dropped, the
    /// session is forgotten and its CID freed, so the next login is not refused with "Session for
    /// CID .. already exists". For a link that died silently (after `probe_server` timed out):
    /// the server's copy is replaced by that login's resume token. Resolves once the session has
    /// dropped; `Err` when this node holds no session for `cid`.
    async fn abandon_session(&self, cid: u64) -> Result<(), NetworkError> {
        crate::server_probe::abandon_session(self, cid).await
    }

    /// Returns all the active sessions in the protocol, including all P2P connections hierarchically placed as children to C2S
    /// connections
    async fn sessions(&self) -> Result<ActiveSessions, NetworkError> {
        match self
            .send_callback_subscription(NodeRequest::GetActiveSessions)
            .await
        {
            Ok(mut stream) => {
                while let Some(result) = stream.next().await {
                    match result.into_result()? {
                        NodeResult::SessionList(res) => return Ok(res.sessions),
                        res => {
                            citadel_logging::warn!("Received unexpected result: {res:?}");
                        }
                    }
                }

                citadel_logging::warn!("Failed to receive response from SDK (stream died)");
                return Err(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteKernelStreamDied,
                    "get_active_sessions"
                ));
            }
            Err(e) => {
                citadel_logging::warn!(target: "citadel", "Failed to query SDK sessions: {e}");
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteQuerySessionsFailed
        ))
    }

    /// Moves this node's live client-role QUIC endpoints to fresh sockets on the current local
    /// address, so Direct and TURN QUIC paths (and a C2S QUIC connection) migrate without a new
    /// handshake. Call it on a local network change. Local only, so not version-gated; see
    /// [`NodeRemote::rebind_local`].
    fn rebind_local(&self) -> Result<RebindReport, NetworkError> {
        self.remote_ref().rebind_local()
    }

    #[doc(hidden)]
    fn remote_ref(&self) -> &NodeRemote<R>;

    #[doc(hidden)]
    async fn get_session_cid<T: Into<UserIdentifier> + Send>(
        &self,
        local_user: T,
    ) -> Result<u64, NetworkError> {
        let account_manager = self.account_manager();
        Ok(account_manager
            .find_local_user_information(local_user)
            .await?
            .ok_or(citadel_io::error!(
                citadel_io::ErrorCode::RemoteUserDoesNotExist
            ))?)
    }
}

impl<R: Ratchet> ProtocolRemoteExt<R> for NodeRemote<R> {
    fn remote_ref(&self) -> &NodeRemote<R> {
        self
    }
}

impl<R: Ratchet> ProtocolRemoteExt<R> for ClientServerRemote<R> {
    fn remote_ref(&self) -> &NodeRemote<R> {
        &self.inner
    }
}

#[async_trait]
/// Some functions require that a target exists
pub trait ProtocolRemoteTargetExt<R: Ratchet>: TargetLockedRemote<R> {
    /// Sends a file with a custom size. The smaller the chunks, the higher the degree of scrambling, but the higher the performance cost. A chunk size of zero will use the default
    async fn send_file_with_custom_opts<T: ObjectSource>(
        &self,
        source: T,
        chunk_size: usize,
        transfer_type: TransferType,
    ) -> Result<(), NetworkError> {
        let chunk_size = if chunk_size == 0 {
            None
        } else {
            Some(chunk_size)
        };
        let session_cid = self.user().get_session_cid();
        let user = *self.user();
        let remote = self.remote();

        let mut stream = remote
            .send_callback_subscription(NodeRequest::SendObject(SendObject {
                source: Box::new(source),
                chunk_size,
                session_cid,
                v_conn_type: user,
                transfer_type,
            }))
            .await?;

        while let Some(event) = stream.next().await {
            match event.into_result()? {
                NodeResult::ObjectTransferHandle(ObjectTransferHandle { mut handle, .. }) => {
                    return handle.transfer_file().await.map_err(|err| {
                        citadel_io::error!(
                            citadel_io::ErrorCode::RemoteFileTransferFailed,
                            err.into_string()
                        )
                    });
                }

                NodeResult::PeerEvent(PeerEvent {
                    event: PeerSignal::SignalReceived { .. },
                    ..
                }) => {}

                res => {
                    log::warn!(target: "citadel", "Invalid NodeResult for FileTransfer request received: {res:?}")
                }
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteFileTransferStreamDied
        ))
    }

    /// Sends a file to the provided target using the default chunking size
    async fn send_file<T: ObjectSource>(&self, source: T) -> Result<(), NetworkError> {
        self.send_file_with_custom_opts(source, 0, TransferType::FileTransfer)
            .await
    }

    /// Sends a file to the provided target using custom chunking size with local encryption.
    /// Only this local node may decrypt the information send to the adjacent node.
    async fn remote_encrypted_virtual_filesystem_push_custom_chunking<
        T: ObjectSource,
        P: Into<PathBuf> + Send,
    >(
        &self,
        source: T,
        virtual_directory: P,
        chunk_size: usize,
        security_level: SecurityLevel,
    ) -> Result<(), NetworkError> {
        self.can_use_revfs()?;
        let mut virtual_path = virtual_directory.into();
        virtual_path = prepare_virtual_path(virtual_path);
        validate_virtual_path(&virtual_path).map_err(|err| {
            citadel_io::error!(
                citadel_io::ErrorCode::RemoteRevfsInvalidVirtualPath,
                err.into_string()
            )
        })?;
        let tx_type = TransferType::RemoteEncryptedVirtualFilesystem {
            virtual_path,
            security_level,
        };
        self.send_file_with_custom_opts(source, chunk_size, tx_type)
            .await
    }

    /// Sends a file to the provided target using the default chunking size with local encryption.
    /// Only this local node may decrypt the information send to the adjacent node.
    async fn remote_encrypted_virtual_filesystem_push<T: ObjectSource, P: Into<PathBuf> + Send>(
        &self,
        source: T,
        virtual_directory: P,
        security_level: SecurityLevel,
    ) -> Result<(), NetworkError> {
        self.remote_encrypted_virtual_filesystem_push_custom_chunking(
            source,
            virtual_directory,
            0,
            security_level,
        )
        .await
    }

    /// Pulls a virtual file from the RE-VFS. If `delete_on_pull` is true, then, the virtual file
    /// will be taken from the RE-VFS
    async fn remote_encrypted_virtual_filesystem_pull<P: Into<PathBuf> + Send>(
        &self,
        virtual_directory: P,
        transfer_security_level: SecurityLevel,
        delete_on_pull: bool,
    ) -> Result<PathBuf, NetworkError> {
        self.can_use_revfs()?;
        let request = NodeRequest::PullObject(PullObject {
            v_conn: *self.user(),
            virtual_dir: virtual_directory.into(),
            delete_on_pull,
            transfer_security_level,
        });

        let mut stream = self.remote().send_callback_subscription(request).await?;

        while let Some(event) = stream.next().await {
            match event.into_result()? {
                NodeResult::ObjectTransferHandle(ObjectTransferHandle { mut handle, .. }) => {
                    return handle.receive_file().await.map_err(|err| {
                        citadel_io::error!(
                            citadel_io::ErrorCode::RemoteFileTransferFailed,
                            err.into_string()
                        )
                    });
                }

                NodeResult::PeerEvent(PeerEvent {
                    event: PeerSignal::SignalReceived { .. },
                    ..
                }) => {}

                // The server's refusal (no such file, or it cannot send it) carries its reason.
                NodeResult::ReVFS(ReVFSResult {
                    error_message: Some(error),
                    ..
                }) => {
                    return Err(citadel_io::error!(
                        citadel_io::ErrorCode::RemoteFileTransferFailed,
                        error
                    ));
                }

                NodeResult::ReVFS(ReVFSResult {
                    error_message: None,
                    ..
                }) => {}

                res => {
                    log::error!(target: "citadel", "Invalid NodeResult for REVFS FileTransfer request received: {res:?}");
                    return Err(citadel_io::error!(
                        citadel_io::ErrorCode::RemoteRevfsInvalidResponse
                    ));
                }
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteRevfsFileTransferStreamDied
        ))
    }

    /// Deletes the file from the RE-VFS. If the contents are desired on delete,
    /// consider calling `Self::remote_encrypted_virtual_filesystem_pull` with the delete
    /// parameter set to true
    async fn remote_encrypted_virtual_filesystem_delete<P: Into<PathBuf> + Send>(
        &self,
        virtual_directory: P,
    ) -> Result<(), NetworkError> {
        self.can_use_revfs()?;
        let request = NodeRequest::DeleteObject(DeleteObject {
            v_conn: *self.user(),
            virtual_dir: virtual_directory.into(),
            security_level: Default::default(),
        });

        let mut stream = self.remote().send_callback_subscription(request).await?;
        while let Some(event) = stream.next().await {
            match event.into_result()? {
                NodeResult::ReVFS(result) => {
                    return if let Some(error) = result.error_message {
                        Err(citadel_io::error!(
                            citadel_io::ErrorCode::RemoteFileTransferFailed,
                            error
                        ))
                    } else {
                        Ok(())
                    }
                }

                evt => {
                    log::error!(target: "citadel", "Invalid NodeResult for REVFS Delete request received: {evt:?}");
                }
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteRevfsDeleteStreamDied
        ))
    }

    /// Connects to the peer with custom settings
    async fn connect_to_peer_custom(
        &self,
        session_security_settings: SessionSecuritySettings,
        udp_mode: UdpMode,
        peer_session_password: Option<PreSharedKey>,
    ) -> Result<PeerConnectSuccess<R>, NetworkError> {
        use std::time::Duration;

        // Bounds the wait for the peer to accept and for the key exchange to finish: the
        // channel is delivered as soon as it works over the server relay, so NAT traversal is no
        // longer inside this wait (it runs in the background; see `PeerChannel::ensure_direct`).
        // What remains is the peer's answer — it has to be online and accept — plus two relayed
        // key-exchange round trips; this prevents an indefinite hang when neither arrives.
        const P2P_CONNECT_TIMEOUT: Duration = Duration::from_secs(60);

        let session_cid = self.user().get_session_cid();
        let peer_target = self.try_as_peer_connection().await?;

        let mut stream = self
            .remote()
            .send_callback_subscription(NodeRequest::PeerCommand(PeerCommand {
                session_cid,
                command: PeerSignal::PostConnect {
                    peer_conn_type: peer_target,
                    ticket_opt: None,
                    invitee_response: None,
                    session_security_settings,
                    udp_mode,
                    session_password: peer_session_password,
                },
            }))
            .await?;

        let connect_task = async {
            while let Some(status) = stream.next().await {
                match status.into_result()? {
                    NodeResult::PeerChannelCreated(PeerChannelCreated {
                        ticket: _,
                        channel,
                        udp_rx_opt,
                    }) => {
                        let username = self.target_username().map(ToString::to_string);
                        let remote = PeerRemote {
                            inner: self.remote().clone(),
                            peer: peer_target.as_virtual_connection(),
                            username,
                            session_security_settings,
                        };

                        return Ok(PeerConnectSuccess {
                            remote,
                            channel: *channel,
                            udp_channel_rx: udp_rx_opt,
                            incoming_object_transfer_handles: None,
                        });
                    }

                    NodeResult::PeerEvent(PeerEvent {
                        event:
                            PeerSignal::PostConnect {
                                invitee_response, ..
                            },
                        ..
                    }) => match invitee_response {
                        Some(PeerResponse::Timeout) => {
                            return Err(citadel_io::error!(
                                citadel_io::ErrorCode::RemotePeerNoResponse
                            ))
                        }
                        Some(PeerResponse::Decline) => {
                            return Err(citadel_io::error!(
                                citadel_io::ErrorCode::RemotePeerDeclined
                            ))
                        }
                        _ => {}
                    },

                    // A routing failure for this ticket (e.g. the server could not deliver our
                    // accept or request). Fail fast with the reason instead of waiting out the
                    // full 60s watchdog — the caller can retry immediately.
                    NodeResult::PeerEvent(PeerEvent {
                        event: PeerSignal::SignalError { error, .. },
                        ..
                    }) => {
                        return Err(citadel_io::error!(
                            citadel_io::ErrorCode::RemoteP2pSignalRoutingFailed,
                            error
                        ))
                    }

                    _ => {}
                }
            }

            Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteKernelStreamDied,
                "connect_to_peer_custom"
            ))
        };

        match citadel_io::time::timeout(P2P_CONNECT_TIMEOUT, connect_task).await {
            Ok(result) => result,
            Err(_elapsed) => Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteP2pConnectTimeout,
                P2P_CONNECT_TIMEOUT.as_secs()
            )),
        }
    }

    /// Supplies (`Some`) or withdraws (`None`) the TURN relay configuration for the next P2P
    /// attempt with this peer. Both peers must supply one for the relay to be used (the
    /// accepting peer before it accepts); the attempt consumes it. When direct P2P is impossible
    /// or fails, the peers then connect through the relay, and
    /// [`PeerChannel::p2p_path`](citadel_proto::prelude::PeerChannel::p2p_path) reports
    /// [`P2pPath::Turn`].
    async fn set_turn_config(
        &self,
        config: Option<citadel_proto::prelude::TurnRelayConfig>,
    ) -> Result<(), NetworkError> {
        let session_cid = self.user().get_session_cid();
        let peer_cid = self
            .try_as_peer_connection()
            .await?
            .get_original_target_cid();
        self.remote()
            .send(NodeRequest::SetPeerTurnConfig(SetPeerTurnConfig {
                session_cid,
                peer_cid,
                config,
            }))
            .await
            .map(|_| ())
    }

    /// [`Self::connect_to_peer_custom`] with a TURN relay configuration for this attempt.
    async fn connect_to_peer_with_turn(
        &self,
        session_security_settings: SessionSecuritySettings,
        udp_mode: UdpMode,
        peer_session_password: Option<PreSharedKey>,
        turn: citadel_proto::prelude::TurnRelayConfig,
    ) -> Result<PeerConnectSuccess<R>, NetworkError> {
        self.set_turn_config(Some(turn)).await?;
        self.connect_to_peer_custom(session_security_settings, udp_mode, peer_session_password)
            .await
    }

    /// Connects to the target peer with default settings
    async fn connect_to_peer(&self) -> Result<PeerConnectSuccess<R>, NetworkError> {
        self.connect_to_peer_custom(Default::default(), Default::default(), Default::default())
            .await
    }

    /// Posts a registration request to a peer
    async fn register_to_peer(&self) -> Result<PeerRegisterStatus, NetworkError> {
        let session_cid = self.user().get_session_cid();
        let peer_target = self.try_as_peer_connection().await?;
        // TODO: Get rid of this step. Should be handled by the protocol
        let local_username = self
            .remote()
            .account_manager()
            .get_username_by_cid(session_cid)
            .await?
            .ok_or_else(|| citadel_io::error!(citadel_io::ErrorCode::RemoteLocalUsernameMissing))?;
        let peer_username_opt = self.target_username().map(ToString::to_string);

        let mut stream = self
            .remote()
            .send_callback_subscription(NodeRequest::PeerCommand(PeerCommand {
                session_cid,
                command: PeerSignal::PostRegister {
                    peer_conn_type: peer_target,
                    inviter_username: local_username,
                    invitee_username: peer_username_opt,
                    ticket_opt: None,
                    invitee_response: None,
                },
            }))
            .await?;

        while let Some(status) = stream.next().await {
            if let NodeResult::PeerEvent(PeerEvent {
                event:
                    PeerSignal::PostRegister {
                        peer_conn_type: _,
                        inviter_username: _,
                        invitee_username: _,
                        ticket_opt: _,
                        invitee_response: Some(resp),
                    },
                ..
            }) = status.into_result()?
            {
                match resp {
                    PeerResponse::Accept(..) => return Ok(PeerRegisterStatus::Accepted),
                    PeerResponse::Decline => return Ok(PeerRegisterStatus::Declined),
                    PeerResponse::Timeout => return Ok(PeerRegisterStatus::Failed { reason: Some("Timeout on register request. Peer did not accept in time. Try again later".to_string()) }),
                    _ => {}
                }
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteKernelStreamDied,
            format!("register_to_peer: {:?}", stream.callback_key())
        ))
    }

    /// Deregisters the currently locked target. If the target is a client to server
    /// connection, deregisters from the server. If the target is a p2p connection,
    /// deregisters the p2p
    async fn deregister(&self) -> Result<(), NetworkError> {
        if let Ok(peer_conn) = self.try_as_peer_connection().await {
            let peer_request = PeerSignal::Deregister {
                peer_conn_type: peer_conn,
            };
            let session_cid = self.user().get_session_cid();
            let request = NodeRequest::PeerCommand(PeerCommand {
                session_cid,
                command: peer_request,
            });

            let mut subscription = self.remote().send_callback_subscription(request).await?;
            while let Some(result) = subscription.next().await {
                if let NodeResult::PeerEvent(PeerEvent {
                    event: PeerSignal::DeregistrationSuccess { .. },
                    ..
                }) = result.into_result()?
                {
                    return Ok(());
                }
            }
        } else {
            // c2s conn
            let cid = self.user().get_session_cid();
            let request = NodeRequest::DeregisterFromHypernode(DeregisterFromHypernode {
                session_cid: cid,
                v_conn_type: *self.user(),
            });
            let mut subscription = self.remote().send_callback_subscription(request).await?;
            while let Some(result) = subscription.next().await {
                match result.into_result()? {
                    NodeResult::DeRegistration(DeRegistration {
                        session_cid: _,
                        ticket_opt: _,
                        success: true,
                    }) => return Ok(()),
                    NodeResult::DeRegistration(DeRegistration {
                        session_cid: _,
                        ticket_opt: _,
                        success: false,
                    }) => {
                        return Err(citadel_io::error!(
                            citadel_io::ErrorCode::RemoteDeregisterFailed
                        ))
                    }

                    _ => {}
                }
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteDeregisterEndedUnexpectedly
        ))
    }

    async fn disconnect(&self) -> Result<(), NetworkError> {
        if let Ok(peer_conn) = self.try_as_peer_connection().await {
            if let PeerConnectionType::LocalGroupPeer {
                session_cid,
                peer_cid: _,
            } = peer_conn
            {
                let request = NodeRequest::PeerCommand(PeerCommand {
                    session_cid,
                    command: PeerSignal::Disconnect {
                        peer_conn_type: peer_conn,
                        disconnect_response: None,
                        disconnect_token: None,
                    },
                });

                let mut subscription = self.remote().send_callback_subscription(request).await?;

                while let Some(event) = subscription.next().await {
                    if let NodeResult::PeerEvent(PeerEvent {
                        event:
                            PeerSignal::Disconnect {
                                peer_conn_type: _,
                                disconnect_response: Some(_),
                                ..
                            },
                        ..
                    }) = event.into_result()?
                    {
                        return Ok(());
                    }
                }

                Err(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteDisconnectEventMissing
                ))
            } else {
                Err(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteExternalGroupPeerUnsupported
                ))
            }
        } else {
            //c2s conn
            let cid = self.user().get_session_cid();
            let request =
                NodeRequest::DisconnectFromHypernode(DisconnectFromHypernode { session_cid: cid });

            let mut subscription = self.remote().send_callback_subscription(request).await?;
            // Bounded, because this wait is otherwise the end of the line.
            //
            // The stream closing is handled below, but a subscription that
            // stays OPEN and simply never carries the matching Disconnect
            // parks here for ever. That is the reconnection suite's 240s
            // timeout: with the phase markers reaching CI, both peers finish
            // phase one and the disconnecting side never logs the line after
            // this call, while the other blocks on the barrier it never
            // reaches. The only await between those two markers is this one.
            //
            // The session is being torn down either way, so an unconfirmed
            // disconnect is reported rather than waited on: a caller can retry
            // or proceed, and neither is possible from inside an indefinite
            // await.
            const DISCONNECT_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(30);
            let deadline = citadel_io::time::Instant::now() + DISCONNECT_CONFIRMATION_TIMEOUT;
            // What the subscription DID carry, for the occasion when it does not
            // carry a Disconnect.
            //
            // `RemoteDisconnectEventMissing` says only that nothing matched in
            // thirty seconds, which cannot distinguish the two things it might
            // mean. A NON-ZERO count says the subscription was alive and other
            // events reached it, so the Disconnect was never emitted or was
            // emitted under a key this subscription does not hold. ZERO says the
            // subscription heard nothing at all, which points at the session
            // rather than at the routing. Those want different fixes, and the
            // error tells them apart not at all.
            //
            // This is the same instrument that localised the UDP one-shot race:
            // an intermittent CI failure that no local run reproduces is only
            // ever going to be solved by what it says about itself when it next
            // happens.
            let mut carried = 0usize;
            loop {
                let remaining =
                    deadline.saturating_duration_since(citadel_io::time::Instant::now());
                if remaining.is_zero() {
                    log::warn!(target: "citadel", "[dc-wait] no Disconnect for cid {cid} within {DISCONNECT_CONFIRMATION_TIMEOUT:?}; the subscription carried {carried} other event(s)");
                    return Err(citadel_io::error!(
                        citadel_io::ErrorCode::RemoteDisconnectEventMissing
                    ));
                }
                match citadel_io::time::timeout(remaining, subscription.next()).await {
                    Ok(Some(event)) => {
                        carried += 1;
                        if let NodeResult::Disconnect(Disconnect {
                            success, message, ..
                        }) = event.into_result()?
                        {
                            return if success {
                                Ok(())
                            } else {
                                Err(citadel_io::error!(
                                    citadel_io::ErrorCode::RemoteDisconnected,
                                    message
                                ))
                            };
                        }
                    }
                    // The stream ended without ever carrying the event.
                    Ok(None) => {
                        log::warn!(target: "citadel", "[dc-wait] the subscription for cid {cid} closed without a Disconnect; it carried {carried} other event(s)");
                        break;
                    }
                    Err(_elapsed) => {
                        log::warn!(target: "citadel", "[dc-wait] timed out waiting for a Disconnect for cid {cid}; the subscription carried {carried} other event(s)");
                        return Err(citadel_io::error!(
                            citadel_io::ErrorCode::RemoteDisconnectEventMissing
                        ));
                    }
                }
            }

            Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteDisconnectEventMissing
            ))
        }
    }

    async fn create_group(
        &self,
        initial_users_to_invite: Option<Vec<UserIdentifier>>,
    ) -> Result<GroupChannel, NetworkError> {
        self.create_group_with_options(initial_users_to_invite, MessageGroupOptions::default())
            .await
    }

    /// Create a group with explicit [`MessageGroupOptions`] — e.g. a zero-trust
    /// [`GroupHierarchyMode::CommandHierarchy`](citadel_types::proto::GroupHierarchyMode) where a
    /// superior can read its subordinates' messages. `options.hierarchy` carries the initial rank
    /// assignment (member cid → command path); it is consumed owner-locally and never reaches the relay.
    async fn create_group_with_options(
        &self,
        initial_users_to_invite: Option<Vec<UserIdentifier>>,
        options: MessageGroupOptions,
    ) -> Result<GroupChannel, NetworkError> {
        let session_cid = self.user().get_session_cid();

        let mut initial_users = vec![];
        // NOTE: default is PRIVATE mode, meaning all users in group must be registered to the owner.
        // Initial users are UserIdentifiers resolved to cids below.
        if let Some(initial_users_to_invite) = initial_users_to_invite {
            for user in initial_users_to_invite {
                initial_users.push(
                    self.remote()
                        .account_manager()
                        .find_target_information(session_cid, user.clone())
                        .await?
                        .ok_or_else(|| {
                            citadel_io::error!(
                                citadel_io::ErrorCode::RemoteGroupAccountNotFound,
                                citadel_io::Dbg(user),
                                citadel_io::Dbg(session_cid)
                            )
                        })
                        .map(|r| r.1.cid)?,
                )
            }
        }

        let group_request = GroupBroadcast::Create {
            initial_invitees: initial_users,
            options,
        };
        let request = NodeRequest::GroupBroadcastCommand(GroupBroadcastCommand {
            session_cid,
            command: group_request,
        });
        let mut subscription = self.remote().send_callback_subscription(request).await?;

        // The loop lives in `group_create_wait` so its refusal paths can be
        // tested without a running node: the only way to make a real server
        // refuse a Create is to give the owner 256 groups first.
        //
        // It matched GroupChannelCreated and nothing else, and — alone among its
        // neighbours — never called into_result(). So SignalError,
        // OutboundRequestRejected and InternalServerError were discarded, and so
        // was the server's own answer to a Create it could not perform:
        // CreateResponse { key: None }, delivered on this very ticket. Every one
        // of those left the caller parked on a stream that would never speak
        // again. The broadcast prefab handles that same event as
        // BroadcastCreateGroupFailed — the guard existed, in the prefab only.
        match crate::group_create_wait::await_group_creation(&mut subscription).await? {
            crate::group_create_wait::GroupCreation::Created(channel) => Ok(channel),
            crate::group_create_wait::GroupCreation::Refused => Err(citadel_io::error!(
                citadel_io::ErrorCode::Generic,
                "The server refused to create the group"
            )),
            crate::group_create_wait::GroupCreation::Ended => Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteCreateGroupEndedUnexpectedly
            )),
        }
    }

    /// Lists all groups that which the current peer owns
    async fn list_owned_groups(&self) -> Result<Vec<MessageGroupKey>, NetworkError> {
        let session_cid = self.user().get_session_cid();
        let cid_to_check_for = match self.try_as_peer_connection().await {
            Ok(res) => res.get_original_target_cid(),
            _ => session_cid,
        };
        let group_request = GroupBroadcast::ListGroupsFor {
            cid: cid_to_check_for,
        };
        let request = NodeRequest::GroupBroadcastCommand(GroupBroadcastCommand {
            session_cid,
            command: group_request,
        });

        let mut subscription = self.remote().send_callback_subscription(request).await?;

        while let Some(evt) = subscription.next().await {
            if let NodeResult::GroupEvent(GroupEvent {
                session_cid: _,
                ticket: _,
                event: GroupBroadcast::ListResponse { groups },
            }) = evt.into_result()?
            {
                return Ok(groups);
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteListGroupsEndedUnexpectedly
        ))
    }

    /// Lists all active sessions, including the local nat type. For each active session,
    /// lists each connection info which includes the remote nat type, connection status,
    /// peer id, and latest ratchet version.
    async fn list_sessions(&self) -> Result<ActiveSessions, NetworkError> {
        let request = NodeRequest::GetActiveSessions;
        let mut subscription = self.remote().send_callback_subscription(request).await?;

        if let Some(NodeResult::SessionList(result)) = subscription.next().await {
            return Ok(result.sessions);
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteListSessionsEndedUnexpectedly
        ))
    }

    /// Begins a re-key, updating the container in the process.
    /// Returns the new key matrix version. Does not return the new key version
    /// if the rekey fails, or, if a current rekey is already executing
    async fn rekey(&self) -> Result<Option<u32>, NetworkError> {
        let request = NodeRequest::ReKey(ReKey {
            v_conn_type: *self.user(),
        });
        let mut subscription = self.remote().send_callback_subscription(request).await?;

        while let Some(evt) = subscription.next().await {
            if let NodeResult::ReKeyResult(result) = evt {
                return match result.status {
                    ReKeyReturnType::Success { version } => Ok(Some(version)),
                    ReKeyReturnType::AlreadyInProgress => Ok(None),
                    ReKeyReturnType::Failure { err } => Err(citadel_io::error!(
                        citadel_io::ErrorCode::RemoteRekeyFailed,
                        err
                    )),
                };
            }
        }

        Err(citadel_io::error!(
            citadel_io::ErrorCode::RemoteRekeyEndedUnexpectedly
        ))
    }

    /// Checks if the locked target is registered
    async fn is_peer_registered(&self) -> Result<bool, NetworkError> {
        let target = self.try_as_peer_connection().await?;
        if let PeerConnectionType::LocalGroupPeer {
            session_cid: local_cid,
            peer_cid,
        } = target
        {
            let peers = self.remote().get_local_group_peers(local_cid, None).await?;
            citadel_logging::info!(target: "citadel", "Checking to see if {target} is registered in {peers:?}");
            Ok(peers.iter().any(|p| p.cid == peer_cid))
        } else {
            Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteExternalGroupPeerUnsupportedYet
            ))
        }
    }

    #[doc(hidden)]
    async fn try_as_peer_connection(&self) -> Result<PeerConnectionType, NetworkError> {
        let verified_return = |user: &VirtualTargetType| {
            user.try_as_peer_connection().ok_or(citadel_io::error!(
                citadel_io::ErrorCode::RemoteTargetNotPeer
            ))
        };

        if self.user().get_target_cid() == 0 {
            // in this case, the user re-used a remote locked to a registration target
            // where the username was provided, but the cid was 0 (unknown).
            let peer_username = self.target_username().ok_or_else(|| {
                citadel_io::error!(citadel_io::ErrorCode::RemoteTargetCidZeroNoUsername)
            })?;
            let session_cid = self.user().get_session_cid();
            let expected_peer_cid = self
                .remote()
                .account_manager()
                .get_persistence_handler()
                .get_cid_by_username(peer_username);
            // get the peer cid from the account manager (implying the peers are already registered).
            // fallback to the mapped cid if the peer is not registered
            let peer_cid = self
                .remote()
                .account_manager()
                .find_target_information(session_cid, peer_username)
                .await?
                .map(|r| r.1.cid)
                .unwrap_or(expected_peer_cid);

            let mut user = *self.user();
            user.set_target_cid(peer_cid);
            verified_return(&user)
        } else {
            verified_return(self.user())
        }
    }

    #[doc(hidden)]
    fn can_use_revfs(&self) -> Result<(), NetworkError> {
        if let Some(sess) = self.session_security_settings() {
            if sess.crypto_params.kem_algorithm == KemAlgorithm::MlKem {
                Ok(())
            } else {
                Err(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteRevfsRequiresKyber
                ))
            }
        } else {
            Err(citadel_io::error!(
                citadel_io::ErrorCode::RemoteRevfsUnsupportedRemote
            ))
        }
    }
}

impl<T: TargetLockedRemote<R>, R: Ratchet> ProtocolRemoteTargetExt<R> for T {}

pub mod results {
    use crate::prefabs::client::peer_connection::FileTransferHandleRx;
    use crate::prelude::{PeerChannel, UdpChannel};
    use crate::remote_ext::remote_specialization::PeerRemote;
    use citadel_io::tokio::sync::oneshot::Receiver;
    use citadel_proto::prelude::*;
    use std::fmt::Debug;

    pub struct PeerConnectSuccess<R: Ratchet> {
        pub channel: PeerChannel<R>,
        pub udp_channel_rx: Option<Receiver<UdpChannel<R>>>,
        pub remote: PeerRemote<R>,
        /// Receives incoming file/object transfer requests. The handles must be
        /// .accepted() before the file/object transfer is allowed to proceed
        pub(crate) incoming_object_transfer_handles: Option<FileTransferHandleRx>,
    }

    impl<R: Ratchet> Debug for PeerConnectSuccess<R> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            f.debug_struct("PeerConnectSuccess")
                .field("channel", &self.channel)
                .field("udp_channel_rx", &self.udp_channel_rx)
                .finish()
        }
    }

    impl<R: Ratchet> PeerConnectSuccess<R> {
        /// Obtains a receiver which yields incoming file/object transfer handles
        pub fn get_incoming_file_transfer_handle(
            &mut self,
        ) -> Result<FileTransferHandleRx, NetworkError> {
            self.incoming_object_transfer_handles
                .take()
                .ok_or(citadel_io::error!(
                    citadel_io::ErrorCode::RemoteFunctionAlreadyCalled
                ))
        }
    }

    /// The peer's answer to a registration request.
    ///
    /// `Ok` means the request completed, not that it was accepted — a decline is
    /// a successful round trip with the answer "no". This type derived nothing,
    /// so a caller could neither compare it nor log it, and all three real
    /// callers wrote `Ok(_)` / `let _ =` and carried on as though the peer had
    /// said yes. `peer_connection.rs` then issued a PostConnect to a peer that
    /// had just declined and waited out a 60s RemoteP2pConnectTimeout, reporting
    /// that instead of the decline the SDK had in hand.
    ///
    /// The derives are the fix for the type; `is_accepted` is the fix for the
    /// call sites, which needed something shorter to write than the mistake.
    #[derive(Clone, Debug, PartialEq, Eq)]
    pub enum PeerRegisterStatus {
        Accepted,
        Declined,
        Failed { reason: Option<String> },
    }

    impl PeerRegisterStatus {
        /// Did the peer say yes?
        pub fn is_accepted(&self) -> bool {
            matches!(self, PeerRegisterStatus::Accepted)
        }

        /// The reason this was not an acceptance, or `None` if it was.
        pub fn refusal_reason(&self) -> Option<String> {
            match self {
                PeerRegisterStatus::Accepted => None,
                PeerRegisterStatus::Declined => Some("The peer declined the request".to_string()),
                PeerRegisterStatus::Failed { reason } => Some(
                    reason
                        .clone()
                        .unwrap_or_else(|| "The registration request failed".to_string()),
                ),
            }
        }
    }

    #[derive(Clone, Debug)]
    pub struct LocalGroupPeer {
        pub cid: u64,
        pub is_online: bool,
    }

    #[derive(Clone, Debug)]
    pub struct LocalGroupPeerFullInfo {
        pub cid: u64,
        pub username: Option<String>,
        pub full_name: Option<String>,
        pub is_online: bool,
    }
}

pub mod remote_specialization {
    use crate::prelude::*;
    use std::ops::{Deref, DerefMut};

    #[derive(Debug, Clone)]
    pub struct PeerRemote<R: Ratchet> {
        pub(crate) inner: NodeRemote<R>,
        pub(crate) peer: VirtualTargetType,
        pub(crate) username: Option<String>,
        pub(crate) session_security_settings: SessionSecuritySettings,
    }

    impl<R: Ratchet> Deref for PeerRemote<R> {
        type Target = NodeRemote<R>;
        fn deref(&self) -> &Self::Target {
            &self.inner
        }
    }

    impl<R: Ratchet> DerefMut for PeerRemote<R> {
        fn deref_mut(&mut self) -> &mut Self::Target {
            &mut self.inner
        }
    }

    impl<R: Ratchet> TargetLockedRemote<R> for PeerRemote<R> {
        fn user(&self) -> &VirtualTargetType {
            &self.peer
        }
        fn remote(&self) -> &NodeRemote<R> {
            &self.inner
        }
        fn target_username(&self) -> Option<&str> {
            self.username.as_deref()
        }
        fn user_mut(&mut self) -> &mut VirtualTargetType {
            &mut self.peer
        }

        fn session_security_settings(&self) -> Option<&SessionSecuritySettings> {
            Some(&self.session_security_settings)
        }
    }
}

#[cfg(all(test, not(target_family = "wasm")))]
mod tests {
    use crate::prefabs::client::single_connection::SingleClientServerConnectionKernel;
    use crate::prefabs::client::DefaultServerConnectionSettingsBuilder;
    use crate::prelude::*;
    use citadel_io::tokio;
    use rstest::rstest;
    use std::net::SocketAddr;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use uuid::Uuid;

    pub struct ReceiverFileTransferKernel<R: Ratchet>(
        pub Option<NodeRemote<R>>,
        pub Arc<AtomicBool>,
    );

    #[async_trait]
    impl<R: Ratchet> NetKernel<R> for ReceiverFileTransferKernel<R> {
        fn load_remote(&mut self, node_remote: NodeRemote<R>) -> Result<(), NetworkError> {
            self.0 = Some(node_remote);
            Ok(())
        }

        async fn on_start(&self) -> Result<(), NetworkError> {
            Ok(())
        }

        async fn on_node_event_received(&self, message: NodeResult<R>) -> Result<(), NetworkError> {
            log::trace!(target: "citadel", "SERVER received {:?}", message);
            if let NodeResult::ObjectTransferHandle(ObjectTransferHandle { mut handle, .. }) =
                message.into_result()?
            {
                let mut path = None;
                // accept the transfer
                handle
                    .accept()
                    .map_err(|err| NetworkError::msg(err.into_string()))?;

                use citadel_types::proto::ObjectTransferStatus;
                use futures::StreamExt;
                while let Some(status) = handle.next().await {
                    match status {
                        ObjectTransferStatus::ReceptionComplete => {
                            log::trace!(target: "citadel", "Server has finished receiving the file!");
                            let cmp = include_bytes!("../../resources/TheBridge.pdf");
                            let streamed_data = citadel_io::tokio::fs::read(path.clone().unwrap())
                                .await
                                .unwrap();
                            assert_eq!(
                                cmp,
                                streamed_data.as_slice(),
                                "Original data and streamed data does not match"
                            );

                            self.1.store(true, Ordering::Relaxed);
                            self.0.clone().unwrap().shutdown().await?;
                        }

                        ObjectTransferStatus::ReceptionBeginning(file_path, vfm) => {
                            path = Some(file_path);
                            assert_eq!(vfm.name, "TheBridge.pdf")
                        }

                        _ => {}
                    }
                }
            }

            Ok(())
        }

        async fn on_stop(&mut self) -> Result<(), NetworkError> {
            Ok(())
        }
    }

    pub fn server_info<'a, R: Ratchet>(
        switch: Arc<AtomicBool>,
    ) -> (NodeFuture<'a, ReceiverFileTransferKernel<R>>, SocketAddr) {
        crate::test_common::server_test_node(ReceiverFileTransferKernel(None, switch), |_| {})
    }

    #[rstest]
    #[case(
        EncryptionAlgorithm::AES_GCM_256,
        KemAlgorithm::MlKem,
        SigAlgorithm::None
    )]
    #[case(
        EncryptionAlgorithm::MlKemHybrid,
        KemAlgorithm::MlKem,
        SigAlgorithm::MlDsa65
    )]
    #[timeout(std::time::Duration::from_secs(90))]
    #[tokio::test]
    async fn test_c2s_file_transfer(
        #[case] enx: EncryptionAlgorithm,
        #[case] kem: KemAlgorithm,
        #[case] sig: SigAlgorithm,
    ) {
        citadel_logging::setup_log();
        let client_success = &AtomicBool::new(false);
        let server_success = &Arc::new(AtomicBool::new(false));
        let (server, server_addr) = server_info::<StackedRatchet>(server_success.clone());
        let uuid = Uuid::new_v4();

        let session_security_settings = SessionSecuritySettingsBuilder::default()
            .with_crypto_params(enx + kem + sig)
            .build()
            .unwrap();

        let server_connection_settings =
            DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
                .with_session_security_settings(session_security_settings)
                .disable_udp()
                .build()
                .unwrap();

        let client_kernel = SingleClientServerConnectionKernel::new(
            server_connection_settings,
            |connection| async move {
                log::trace!(target: "citadel", "***CLIENT LOGIN SUCCESS :: File transfer next ***");
                connection
                    .send_file_with_custom_opts(
                        "../resources/TheBridge.pdf",
                        32 * 1024,
                        TransferType::FileTransfer,
                    )
                    .await
                    .unwrap();
                log::trace!(target: "citadel", "***CLIENT FILE TRANSFER SUCCESS***");
                client_success.store(true, Ordering::Relaxed);
                connection.shutdown_kernel().await
            },
        );

        let client = DefaultNodeBuilder::default()
            .with_stun_servers(crate::test_common::LOCALHOST_STUN_SERVERS)
            .build(client_kernel)
            .unwrap();

        let joined = futures::future::try_join(server, client);

        let _ = joined.await.unwrap();

        assert!(client_success.load(Ordering::Relaxed));
        assert!(server_success.load(Ordering::Relaxed));
    }
}