freenet 0.2.48

//! Network messaging between peers.
//! Defines the `NetMessage` enum, the standard format for all peer-to-peer communication within the Freenet network.
//! See `architecture.md`.

#[cfg(feature = "trace-ot")]
use std::time::SystemTime;
use std::{borrow::Cow, fmt::Display, net::SocketAddr, time::Duration};

use crate::{
    client_events::{ClientId, HostResult},
    operations::{
        connect::ConnectMsg, get::GetMsg, put::PutMsg, subscribe::SubscribeMsg, update::UpdateMsg,
    },
    ring::{Location, PeerKeyLocation},
};
use freenet_stdlib::prelude::{
    ContractContainer, ContractInstanceId, ContractKey, DelegateKey, WrappedState,
};
pub(crate) use sealed_msg_type::{TransactionType, TransactionTypeId};
use serde::{Deserialize, Serialize};
use ulid::Ulid;

/// An transaction is a unique, universal and efficient identifier for any
/// roundtrip transaction as it is broadcasted around the Freenet network.
///
/// The identifier conveys all necessary information to identify and classify the
/// transaction:
/// - The unique identifier itself.
/// - The type of transaction being performed.
/// - If the transaction has been finalized, this allows for the connection manager
///   to sweep any garbage left by a finished (or timed out) transaction.
///
/// A transaction may span different messages sent across the network.
#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy)]
pub struct Transaction {
    id: Ulid,
    /// Parent transaction ID for child operations spawned by this transaction.
    /// Enables atomicity tracking for composite operations (e.g., PUT with SUBSCRIBE).
    parent: Option<Ulid>,
}

impl Transaction {
    pub const NULL: &'static Transaction = &Transaction {
        id: Ulid(0),
        parent: None,
    };

    pub(crate) fn new<T: TxType>() -> Self {
        let ty = <T as TxType>::tx_type_id();
        let id = crate::config::GlobalSimulationTime::new_ulid();
        Self::update(ty.0, id, None)
    }

    /// Creates a child transaction with the specified type, linked to the parent
    /// for atomicity tracking in composite operations.
    pub(crate) fn new_child_of<T: TxType>(parent: &Transaction) -> Self {
        let ty = <T as TxType>::tx_type_id();
        let id = crate::config::GlobalSimulationTime::new_ulid();
        Self::update(ty.0, id, Some(parent.id))
    }

    /// Returns the parent transaction ID for child operations.
    pub fn parent_id(&self) -> Option<&Ulid> {
        self.parent.as_ref()
    }

    /// Returns true if this transaction is a child operation.
    pub fn is_sub_operation(&self) -> bool {
        self.parent.is_some()
    }

    pub(crate) fn transaction_type(&self) -> TransactionType {
        let id_byte = (self.id.0 & 0xFFu128) as u8;
        TransactionType::try_from(id_byte).expect(
            "Transaction ID contains invalid type byte; this is a bug in Transaction construction",
        )
    }

    pub fn timed_out(&self) -> bool {
        self.elapsed() >= crate::config::OPERATION_TTL
    }

    #[cfg(feature = "trace-ot")]
    pub fn started(&self) -> SystemTime {
        SystemTime::UNIX_EPOCH + Duration::from_millis(self.id.timestamp_ms())
    }

    #[cfg(feature = "trace-ot")]
    pub fn as_bytes(&self) -> [u8; 16] {
        self.id.0.to_le_bytes()
    }

    /// Returns the transaction ID as raw bytes.
    /// Used for deriving hash keys in bloom filters.
    pub fn id_bytes(&self) -> [u8; 16] {
        self.id.0.to_le_bytes()
    }

    /// Returns the elapsed time since this transaction was created.
    ///
    /// Uses simulation time when in simulation mode, otherwise system time.
    /// This ensures deterministic elapsed time calculations in DST tests.
    pub fn elapsed(&self) -> Duration {
        use crate::config::GlobalSimulationTime;
        let current_unix_epoch_ts = GlobalSimulationTime::read_time_ms();
        let this_tx_creation = self.id.timestamp_ms();
        if current_unix_epoch_ts < this_tx_creation {
            Duration::new(0, 0)
        } else {
            let ms_elapsed = current_unix_epoch_ts - this_tx_creation;
            Duration::from_millis(ms_elapsed)
        }
    }

    /// Generate a random transaction which has the implicit TTL cutoff.
    ///
    /// This will allow, for example, to compare against any older transactions,
    /// in order to remove them.
    pub fn ttl_transaction() -> Self {
        Self::ttl_transaction_with_multiplier(1)
    }

    /// Like [`ttl_transaction`](Self::ttl_transaction) but with a custom TTL multiplier.
    ///
    /// Used for absolute timeout enforcement on operations that would otherwise
    /// be exempt from garbage collection (e.g., `under_progress` operations).
    pub fn ttl_transaction_with_multiplier(multiplier: u64) -> Self {
        let id = crate::config::GlobalSimulationTime::new_ulid();
        let ts = id.timestamp_ms();
        let ttl_ms = crate::config::OPERATION_TTL.as_millis() as u64 * multiplier;
        let ttl_epoch: u64 = ts.saturating_sub(ttl_ms);

        // Clear the timestamp bits and replace with the cutoff timestamp.
        const TIMESTAMP_MASK: u128 = 0x00000000000000000000FFFFFFFFFFFFFFFF;
        let new_ulid = (id.0 & TIMESTAMP_MASK) | ((ttl_epoch as u128) << 80);
        Self {
            id: Ulid(new_ulid),
            parent: None,
        }
    }

    fn update(ty: TransactionType, id: Ulid, parent: Option<Ulid>) -> Self {
        const TYPE_MASK: u128 = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00u128;
        // Clear the last byte
        let cleared = id.0 & TYPE_MASK;
        // Set the last byte with the transaction type
        let updated = cleared | (ty as u8) as u128;

        // 2 words size for 64-bits platforms
        Self {
            id: Ulid(updated),
            parent,
        }
    }
}

#[cfg(test)]
impl<'a> arbitrary::Arbitrary<'a> for Transaction {
    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
        let ty: TransactionTypeId = u.arbitrary()?;
        let bytes: u128 = Ulid::new().0;
        Ok(Self::update(ty.0, Ulid(bytes), None))
    }
}

impl Display for Transaction {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.id)
    }
}

impl std::fmt::Debug for Transaction {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.id)
    }
}

impl PartialOrd for Transaction {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for Transaction {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.id.cmp(&other.id)
    }
}

/// Get the transaction type associated to a given message type.
pub trait TxType: sealed_msg_type::SealedTxType {
    fn tx_type_id() -> TransactionTypeId;
}

impl<T> TxType for T
where
    T: sealed_msg_type::SealedTxType,
{
    fn tx_type_id() -> TransactionTypeId {
        <Self as sealed_msg_type::SealedTxType>::tx_type_id()
    }
}

mod sealed_msg_type {
    use super::*;
    use crate::operations::connect::ConnectMsg;

    pub trait SealedTxType {
        fn tx_type_id() -> TransactionTypeId;
    }

    #[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
    #[cfg_attr(test, derive(arbitrary::Arbitrary))]
    pub struct TransactionTypeId(pub(super) TransactionType);

    #[repr(u8)]
    #[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
    #[cfg_attr(test, derive(arbitrary::Arbitrary))]
    pub enum TransactionType {
        Connect = 0,
        Put = 1,
        Get = 2,
        Subscribe = 3,
        Update = 4,
    }

    impl TryFrom<u8> for TransactionType {
        type Error = u8;

        fn try_from(value: u8) -> Result<Self, Self::Error> {
            match value {
                0 => Ok(TransactionType::Connect),
                1 => Ok(TransactionType::Put),
                2 => Ok(TransactionType::Get),
                3 => Ok(TransactionType::Subscribe),
                4 => Ok(TransactionType::Update),
                other => Err(other),
            }
        }
    }

    impl TransactionType {
        pub fn description(&self) -> &'static str {
            match self {
                TransactionType::Connect => "connect",
                TransactionType::Put => "put",
                TransactionType::Get => "get",
                TransactionType::Subscribe => "subscribe",
                TransactionType::Update => "update",
            }
        }
    }

    impl Display for TransactionType {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "{}", self.description())
        }
    }

    macro_rules! transaction_type_enumeration {
        ($variant:ident, $enum_type:ident, decl struct { $( $var:ident -> $ty:ty ),+ }) => {
            $(
                impl From<$ty> for NetMessage {
                    fn from(msg: $ty) -> Self {
                        Self::$variant($enum_type::$var(msg))
                    }
                }

                impl SealedTxType for $ty {
                    fn tx_type_id() -> TransactionTypeId {
                        TransactionTypeId(TransactionType::$var)
                    }
                }
            )+
        };
    }

    transaction_type_enumeration!(V1, NetMessageV1, decl struct {
        Connect -> ConnectMsg,
        Put -> PutMsg,
        Get -> GetMsg,
        Subscribe -> SubscribeMsg,
        Update -> UpdateMsg
    });
}

pub(crate) trait MessageStats {
    fn id(&self) -> &Transaction;

    fn requested_location(&self) -> Option<Location>;
}

/// Wrapper for inbound messages that carries the source address from the transport layer.
/// This separates routing concerns from message content - the source address is determined by
/// the network layer (from the packet), not embedded in the serialized message.
///
/// Generic over the message type so it can wrap:
/// - `NetMessage` at the network layer (p2p_protoc.rs)
/// - Specific operation messages (GetMsg, PutMsg, etc.) at the operation layer
///
/// Note: Currently unused but prepared for Phase 4 of #2164.
/// Will be used to thread source addresses to operations for routing.
#[allow(dead_code)]
#[derive(Debug, Clone)]
pub struct InboundMessage<M> {
    /// The message content
    pub msg: M,
    /// The socket address this message was received from (from UDP packet source)
    pub source_addr: SocketAddr,
}

#[allow(dead_code)]
impl<M> InboundMessage<M> {
    /// Create a new inbound message wrapper
    pub fn new(msg: M, source_addr: SocketAddr) -> Self {
        Self { msg, source_addr }
    }

    /// Transform the inner message while preserving source_addr
    pub fn map<N>(self, f: impl FnOnce(M) -> N) -> InboundMessage<N> {
        InboundMessage {
            msg: f(self.msg),
            source_addr: self.source_addr,
        }
    }

    /// Get a reference to the inner message
    pub fn inner(&self) -> &M {
        &self.msg
    }
}

#[allow(dead_code)]
impl InboundMessage<NetMessage> {
    /// Get the transaction ID from the wrapped network message
    pub fn id(&self) -> &Transaction {
        self.msg.id()
    }
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub(crate) enum NetMessage {
    V1(NetMessageV1),
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub(crate) enum NetMessageV1 {
    Connect(ConnectMsg),
    Put(PutMsg),
    Get(GetMsg),
    Subscribe(SubscribeMsg),
    Update(UpdateMsg),
    Aborted(Transaction),
    /// Neighbor hosting protocol message for tracking which neighbors host which contracts.
    NeighborHosting {
        message: NeighborHostingMessage,
    },
    /// Interest synchronization protocol for delta-based updates.
    InterestSync {
        message: InterestMessage,
    },
    /// Peer readiness advertisement: indicates whether the sender is ready
    /// to accept non-CONNECT operations. Peers behind symmetric NAT with
    /// only a gateway connection broadcast `ready: false` (implicitly, by
    /// not yet sending this message) and `ready: true` once they have
    /// enough ring connections (`min_ready_connections`).
    ReadyState {
        ready: bool,
    },
}

/// Messages for the neighbor hosting protocol.
///
/// This protocol allows neighbors to inform each other which contracts they are hosting,
/// enabling UPDATE forwarding to hosts who may not be explicitly subscribed.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(clippy::enum_variant_names)]
pub enum NeighborHostingMessage {
    /// Announce changes to our hosted contracts.
    HostingAnnounce {
        /// Contracts we've started hosting.
        added: Vec<ContractInstanceId>,
        /// Contracts we've stopped hosting.
        removed: Vec<ContractInstanceId>,
        /// True if this is a response to a received announcement.
        /// Recipients should not respond to responses (prevents ping-pong).
        #[serde(default)]
        is_response: bool,
    },
    /// Request a neighbor's full hosting state (used on new connections).
    HostingStateRequest,
    /// Response with the neighbor's full hosting state.
    HostingStateResponse { contracts: Vec<ContractInstanceId> },
}

/// Messages for the delta-based interest synchronization protocol.
///
/// This protocol enables peers to:
/// 1. Discover shared contract interests at connection time
/// 2. Exchange state summaries for delta computation
/// 3. Track interest changes during the connection
/// 4. Request full state resync when delta application fails
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum InterestMessage {
    /// Connection-time interest exchange.
    ///
    /// Sent by both peers immediately after connection establishment.
    /// Contains fast hashes (FNV-1a) of contract IDs for efficient matching.
    Interests {
        /// Fast u32 hashes of contract IDs we're interested in.
        hashes: Vec<u32>,
    },

    /// State summaries for contracts both peers share interest in.
    ///
    /// Sent after comparing `Interests` hashes. Only includes summaries
    /// for contracts where we have state (summary is None if we have no state).
    Summaries {
        /// (contract_hash, summary_bytes) pairs for shared contracts.
        /// Summary bytes is None if we're interested but don't have state yet.
        /// Use `SummaryEntry::from_summary()` to create entries.
        entries: Vec<SummaryEntry>,
    },

    /// Incremental changes to our contract interests.
    ///
    /// Sent when we gain or lose interest in contracts after connection.
    /// The receiver responds with Summaries for newly shared contracts.
    ChangeInterests {
        /// Contract hashes we've newly become interested in.
        added: Vec<u32>,
        /// Contract hashes we're no longer interested in.
        removed: Vec<u32>,
    },

    /// Request full state resync when delta application fails.
    ///
    /// Sent when a received delta cannot be applied (corruption, version mismatch).
    /// The upstream peer responds with ResyncResponse containing full state.
    ResyncRequest {
        /// The contract that needs resync.
        key: ContractKey,
    },

    /// Response to ResyncRequest with full state.
    ///
    /// Sent when a peer requests resync after delta application failure.
    /// Contains the full contract state and sender's summary.
    ResyncResponse {
        /// The contract being resynced.
        key: ContractKey,
        /// Full contract state bytes.
        state_bytes: Vec<u8>,
        /// Sender's current state summary bytes.
        summary_bytes: Vec<u8>,
    },
}

/// A summary entry for the Summaries message.
/// Uses owned bytes for wire serialization.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SummaryEntry {
    /// Fast hash of the contract ID.
    pub hash: u32,
    /// Summary bytes, or None if we're interested but don't have state yet.
    pub summary_bytes: Option<Vec<u8>>,
}

impl SummaryEntry {
    /// Create a summary entry from a contract hash and optional summary.
    pub fn from_summary(
        hash: u32,
        summary: Option<&freenet_stdlib::prelude::StateSummary<'_>>,
    ) -> Self {
        Self {
            hash,
            summary_bytes: summary.map(|s| s.as_ref().to_vec()),
        }
    }

    /// Convert the summary bytes back to a StateSummary.
    pub fn to_summary(&self) -> Option<freenet_stdlib::prelude::StateSummary<'static>> {
        self.summary_bytes
            .as_ref()
            .map(|bytes| freenet_stdlib::prelude::StateSummary::from(bytes.clone()))
    }
}

/// Payload for delta-based updates.
///
/// Used in update messages to send either a delta (when we know the peer's summary)
/// or full state (when we don't know their state or delta would be inefficient).
///
/// NOTE: This type provides foundation infrastructure for delta-based updates.
/// Methods are marked `#[allow(dead_code)]` because they will be used in
/// follow-up PRs that integrate the full delta sync workflow.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(dead_code)]
pub enum DeltaOrFullState {
    /// A delta computed from the peer's cached summary.
    /// More efficient when both peers have state and the delta is small.
    /// Uses owned bytes for wire serialization.
    Delta(Vec<u8>),

    /// Full contract state. Used when:
    ///
    /// - Peer has no cached summary (first sync)
    /// - Delta would be larger than 50% of state size
    /// - After a ResyncRequest
    ///
    /// Uses owned bytes for wire serialization.
    FullState(Vec<u8>),
}

#[allow(dead_code)]
impl DeltaOrFullState {
    /// Create a Delta variant from a StateDelta.
    pub fn from_delta(delta: &freenet_stdlib::prelude::StateDelta<'_>) -> Self {
        Self::Delta(delta.as_ref().to_vec())
    }

    /// Create a FullState variant from a State.
    pub fn from_state(state: &freenet_stdlib::prelude::State<'_>) -> Self {
        Self::FullState(state.as_ref().to_vec())
    }

    /// Convert to a StateDelta (if this is a Delta variant).
    pub fn to_delta(&self) -> Option<freenet_stdlib::prelude::StateDelta<'static>> {
        match self {
            Self::Delta(bytes) => Some(freenet_stdlib::prelude::StateDelta::from(bytes.clone())),
            Self::FullState(_) => None,
        }
    }

    /// Convert to a State (if this is a FullState variant).
    pub fn to_state(&self) -> Option<freenet_stdlib::prelude::State<'static>> {
        match self {
            Self::Delta(_) => None,
            Self::FullState(bytes) => Some(freenet_stdlib::prelude::State::from(bytes.clone())),
        }
    }

    /// Check if this is a delta (not full state).
    pub fn is_delta(&self) -> bool {
        matches!(self, Self::Delta(_))
    }

    /// Get the raw bytes of the payload.
    fn bytes(&self) -> &[u8] {
        match self {
            Self::Delta(bytes) | Self::FullState(bytes) => bytes,
        }
    }

    /// Get the size in bytes of the payload.
    pub fn size(&self) -> usize {
        self.bytes().len()
    }
}

trait Versioned {
    fn version(&self) -> semver::Version;
}

impl Versioned for NetMessage {
    fn version(&self) -> semver::Version {
        match self {
            NetMessage::V1(inner) => inner.version(),
        }
    }
}

impl Versioned for NetMessageV1 {
    fn version(&self) -> semver::Version {
        match self {
            NetMessageV1::Connect(_) => semver::Version::new(1, 1, 0),
            NetMessageV1::Put(_) => semver::Version::new(1, 0, 0),
            NetMessageV1::Get(_) => semver::Version::new(1, 1, 0),
            NetMessageV1::Subscribe(_) => semver::Version::new(1, 1, 0),
            // Version 2.0.0 for delta-based BroadcastTo format
            NetMessageV1::Update(_) => semver::Version::new(2, 0, 0),
            NetMessageV1::Aborted(_) => semver::Version::new(1, 0, 0),
            NetMessageV1::NeighborHosting { .. } => semver::Version::new(1, 0, 0),
            // Version 1.1.0 for delta-based interest sync
            NetMessageV1::InterestSync { .. } => semver::Version::new(1, 1, 0),
            NetMessageV1::ReadyState { .. } => semver::Version::new(1, 2, 0),
        }
    }
}

impl From<NetMessage> for semver::Version {
    fn from(msg: NetMessage) -> Self {
        msg.version()
    }
}

pub trait InnerMessage: Into<NetMessage> {
    fn id(&self) -> &Transaction;

    fn requested_location(&self) -> Option<Location>;
}

type RemainingChecks = Option<usize>;
type ConnectResult = Result<(SocketAddr, RemainingChecks), ()>;

/// Internal node events emitted to the event loop.
#[derive(Debug, Clone)]
pub(crate) enum NodeEvent {
    /// Drop the given peer connection by socket address.
    DropConnection(SocketAddr),
    /// Drop all connections (ring + transient). Used after suspend/resume
    /// to force fresh transport reconnection to gateways.
    DropAllConnections,
    // Try connecting to the given peer.
    ConnectPeer {
        peer: PeerKeyLocation,
        tx: Transaction,
        callback: tokio::sync::mpsc::Sender<ConnectResult>,
        is_gw: bool,
    },
    Disconnect {
        cause: Option<Cow<'static, str>>,
    },
    QueryConnections {
        callback: tokio::sync::mpsc::Sender<QueryResult>,
    },
    QuerySubscriptions {
        callback: tokio::sync::mpsc::Sender<QueryResult>,
    },
    QueryNodeDiagnostics {
        config: freenet_stdlib::client_api::NodeDiagnosticsConfig,
        callback: tokio::sync::mpsc::Sender<QueryResult>,
    },
    TransactionTimedOut(Transaction),
    /// Transaction completed successfully - cleanup client subscription
    TransactionCompleted(Transaction),
    /// **Standalone** subscription completed - deliver SubscribeResponse to client via result router.
    ///
    /// **IMPORTANT:** This event is ONLY used for standalone subscriptions (no remote peers available).
    /// Normal network subscriptions go through `handle_op_result`, which sends results via
    /// `result_router_tx` directly without needing this event.
    ///
    /// **Architecture Note (Issue #2075):**
    /// Local client subscriptions are handled separately from network peer subscriptions:
    /// - Subsequent contract updates are delivered via the executor's `update_notifications`
    ///   channels (see `send_update_notification` in runtime.rs)
    /// - Network peer subscriptions use the `hosting_manager.subscribers` for UPDATE propagation
    LocalSubscribeComplete {
        tx: Transaction,
        key: ContractKey,
        subscribed: bool,
        /// Whether this was a node-internal subscription renewal (no client waiting).
        is_renewal: bool,
    },
    /// Register expectation for an inbound connection from the given peer.
    ExpectPeerConnection {
        addr: SocketAddr,
    },
    /// Broadcast a proximity cache message to all connected peers.
    BroadcastHostingUpdate {
        message: NeighborHostingMessage,
    },
    /// Broadcast a ChangeInterests message to all connected peers for delta sync.
    BroadcastChangeInterests {
        added: Vec<u32>,
        removed: Vec<u32>,
    },
    /// Send an interest message to a specific peer.
    /// Used for ResyncRequest when delta application fails.
    SendInterestMessage {
        target: SocketAddr,
        message: InterestMessage,
    },
    /// Broadcast state change to interested network peers.
    /// Emitted by executor when local state changes.
    /// Handled by p2p_protoc which has access to OpManager and network.
    BroadcastStateChange {
        key: ContractKey,
        new_state: WrappedState,
    },
    /// Send state to a specific peer that reported a stale summary.
    /// Unlike BroadcastStateChange (which fans out to ALL subscribers),
    /// this targets only the peer that needs catching up.
    SyncStateToPeer {
        key: ContractKey,
        new_state: WrappedState,
        target: SocketAddr,
    },
}

#[derive(Debug, Clone)]
pub struct SubscriptionInfo {
    pub instance_id: ContractInstanceId,
    pub client_id: ClientId,
    pub last_update: Option<std::time::SystemTime>,
}

#[derive(Debug, Clone)]
pub struct NetworkDebugInfo {
    /// Application-level subscriptions (WebSocket clients subscribed to contracts)
    pub application_subscriptions: Vec<SubscriptionInfo>,
    /// Network-level subscriptions (nodes subscribing to contracts for routing)
    #[allow(dead_code)] // Used for debugging purposes, not exposed via stdlib API yet
    pub network_subscriptions: Vec<(ContractKey, Vec<SocketAddr>)>,
    pub connected_peers: Vec<PeerKeyLocation>,
}

#[derive(Debug)]
pub(crate) enum QueryResult {
    Connections(Vec<PeerKeyLocation>),
    GetResult {
        key: ContractKey,
        state: WrappedState,
        contract: Option<ContractContainer>,
    },
    DelegateResult {
        #[allow(dead_code)]
        key: DelegateKey,
        response: HostResult,
    },
    NetworkDebug(NetworkDebugInfo),
    NodeDiagnostics(freenet_stdlib::client_api::NodeDiagnosticsResponse),
}

impl Display for NodeEvent {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            NodeEvent::DropConnection(peer) => {
                write!(f, "DropConnection (from {peer})")
            }
            NodeEvent::DropAllConnections => {
                write!(f, "DropAllConnections")
            }
            NodeEvent::ConnectPeer { peer, .. } => {
                write!(f, "ConnectPeer (to {peer})")
            }
            NodeEvent::Disconnect { cause: Some(cause) } => {
                write!(f, "Disconnect node, reason: {cause}")
            }
            NodeEvent::Disconnect { cause: None } => {
                write!(f, "Disconnect node, reason: unknown")
            }
            NodeEvent::QueryConnections { .. } => {
                write!(f, "QueryConnections")
            }
            NodeEvent::QuerySubscriptions { .. } => {
                write!(f, "QuerySubscriptions")
            }
            NodeEvent::QueryNodeDiagnostics { .. } => {
                write!(f, "QueryNodeDiagnostics")
            }
            NodeEvent::TransactionTimedOut(transaction) => {
                write!(f, "Transaction timed out ({transaction})")
            }
            NodeEvent::TransactionCompleted(transaction) => {
                write!(f, "Transaction completed ({transaction})")
            }
            NodeEvent::LocalSubscribeComplete {
                tx,
                key,
                subscribed,
                ..
            } => {
                write!(
                    f,
                    "Local subscribe complete (tx: {tx}, key: {key}, subscribed: {subscribed})"
                )
            }
            NodeEvent::ExpectPeerConnection { addr } => {
                write!(f, "ExpectPeerConnection (from {addr})")
            }
            NodeEvent::BroadcastHostingUpdate { message } => {
                write!(f, "BroadcastHostingUpdate ({message:?})")
            }
            NodeEvent::BroadcastChangeInterests { added, removed } => {
                write!(
                    f,
                    "BroadcastChangeInterests (added: {}, removed: {})",
                    added.len(),
                    removed.len()
                )
            }
            NodeEvent::SendInterestMessage { target, message } => {
                let msg_summary = match message {
                    InterestMessage::Interests { hashes } => {
                        format!("Interests({} hashes)", hashes.len())
                    }
                    InterestMessage::Summaries { entries } => {
                        format!("Summaries({} entries)", entries.len())
                    }
                    InterestMessage::ChangeInterests { added, removed } => {
                        format!(
                            "ChangeInterests(+{} -{} hashes)",
                            added.len(),
                            removed.len()
                        )
                    }
                    InterestMessage::ResyncRequest { key } => {
                        format!("ResyncRequest({key})")
                    }
                    InterestMessage::ResyncResponse {
                        key, state_bytes, ..
                    } => {
                        format!("ResyncResponse({key}, {} bytes)", state_bytes.len())
                    }
                };
                write!(f, "SendInterestMessage (to: {target}, {msg_summary})")
            }
            NodeEvent::BroadcastStateChange { key, .. } => {
                write!(f, "BroadcastStateChange (contract: {key})")
            }
            NodeEvent::SyncStateToPeer { key, target, .. } => {
                write!(f, "SyncStateToPeer (contract: {key}, target: {target})")
            }
        }
    }
}

impl MessageStats for NetMessage {
    fn id(&self) -> &Transaction {
        match self {
            NetMessage::V1(msg) => msg.id(),
        }
    }

    fn requested_location(&self) -> Option<Location> {
        match self {
            NetMessage::V1(msg) => msg.requested_location(),
        }
    }
}

impl MessageStats for NetMessageV1 {
    fn id(&self) -> &Transaction {
        match self {
            NetMessageV1::Connect(op) => op.id(),
            NetMessageV1::Put(op) => op.id(),
            NetMessageV1::Get(op) => op.id(),
            NetMessageV1::Subscribe(op) => op.id(),
            NetMessageV1::Update(op) => op.id(),
            NetMessageV1::Aborted(tx) => tx,
            NetMessageV1::NeighborHosting { .. } => Transaction::NULL,
            NetMessageV1::InterestSync { .. } => Transaction::NULL,
            NetMessageV1::ReadyState { .. } => Transaction::NULL,
        }
    }

    fn requested_location(&self) -> Option<Location> {
        match self {
            NetMessageV1::Connect(op) => op.requested_location(),
            NetMessageV1::Put(op) => op.requested_location(),
            NetMessageV1::Get(op) => op.requested_location(),
            NetMessageV1::Subscribe(op) => op.requested_location(),
            NetMessageV1::Update(op) => op.requested_location(),
            NetMessageV1::Aborted(_) => None,
            NetMessageV1::NeighborHosting { .. } => None,
            NetMessageV1::InterestSync { .. } => None,
            NetMessageV1::ReadyState { .. } => None,
        }
    }
}

impl Display for NetMessage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        use NetMessageV1::*;
        write!(f, "Message {{")?;
        match self {
            NetMessage::V1(msg) => match msg {
                Connect(msg) => msg.fmt(f)?,
                Put(msg) => msg.fmt(f)?,
                Get(msg) => msg.fmt(f)?,
                Subscribe(msg) => msg.fmt(f)?,
                Update(msg) => msg.fmt(f)?,
                Aborted(msg) => msg.fmt(f)?,
                NeighborHosting { message } => {
                    write!(f, "NeighborHosting {{ {message:?} }}")?;
                }
                InterestSync { message } => {
                    write!(f, "InterestSync {{ {message:?} }}")?;
                }
                ReadyState { ready } => {
                    write!(f, "ReadyState {{ ready: {ready} }}")?;
                }
            },
        };
        write!(f, "}}")
    }
}

// ── Compile-time invariant checks ──────────────────────────────────────
//
// These const assertions catch layout and enum-variant assumptions at
// compile time, preventing a whole class of bugs that previously could
// only surface at runtime (or worse, as UB via unreachable_unchecked).

/// Transaction layout: Ulid (16 bytes) + Option<Ulid> (24 bytes, with niche) = 40 bytes.
/// Any change to this layout would break serialization compatibility and network protocol.
const _: () = {
    // Ulid is a newtype over u128 (16 bytes).
    assert!(std::mem::size_of::<ulid::Ulid>() == 16, "Ulid size changed");
    // Transaction = { id: Ulid, parent: Option<Ulid> }.
    // Assert it stays within a reasonable bound (≤48 bytes).
    assert!(
        std::mem::size_of::<Transaction>() <= 48,
        "Transaction size grew beyond expected bounds — check serialization compatibility"
    );
};

/// TransactionType must have exactly 5 variants (0..=4).
/// If a new variant is added, `Transaction::transaction_type()` and the
/// `TryFrom<u8>` impl must be updated, and this assertion bumped.
const _: () = {
    // The highest valid discriminant must be 4 (Update).
    assert!(
        sealed_msg_type::TransactionType::Update as u8 == 4,
        "TransactionType variants changed — update TryFrom<u8> and this assertion"
    );
};

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn pack_transaction_type() {
        let ts_0 = Ulid::new();
        std::thread::sleep(Duration::from_millis(1));
        let tx = Transaction::update(TransactionType::Connect, Ulid::new(), None);
        assert_eq!(tx.transaction_type(), TransactionType::Connect);
        let tx = Transaction::update(TransactionType::Subscribe, Ulid::new(), None);
        assert_eq!(tx.transaction_type(), TransactionType::Subscribe);
        std::thread::sleep(Duration::from_millis(1));
        let ts_1 = Ulid::new();
        assert!(
            tx.id.timestamp_ms() > ts_0.timestamp_ms(),
            "{:?} <= {:?}",
            tx.id.datetime(),
            ts_0.datetime()
        );
        assert!(
            tx.id.timestamp_ms() < ts_1.timestamp_ms(),
            "{:?} >= {:?}",
            tx.id.datetime(),
            ts_1.datetime()
        );
    }

    #[test]
    fn get_ttl_cutoff_transaction() {
        let ttl_tx = Transaction::ttl_transaction();
        let original_tx = Transaction::new::<crate::operations::get::GetMsg>();

        assert!(original_tx > ttl_tx);
        assert!(ttl_tx.timed_out());
        assert!(
            original_tx.id.timestamp_ms() - ttl_tx.id.timestamp_ms()
                >= crate::config::OPERATION_TTL.as_millis() as u64
        );
        assert!(
            original_tx.id.timestamp_ms() - ttl_tx.id.timestamp_ms()
                < crate::config::OPERATION_TTL.as_millis() as u64 + 5
        );
    }

    #[test]
    fn ttl_transaction_with_multiplier_produces_older_cutoff() {
        let ttl_1x = Transaction::ttl_transaction();
        let ttl_5x = Transaction::ttl_transaction_with_multiplier(5);

        // 5x multiplier should produce an older (smaller timestamp) cutoff
        assert!(ttl_5x < ttl_1x, "5x multiplier should be older than 1x");

        // Verify the timestamp delta is approximately 4x OPERATION_TTL more
        let diff = ttl_1x.id.timestamp_ms() - ttl_5x.id.timestamp_ms();
        let expected = crate::config::OPERATION_TTL.as_millis() as u64 * 4;
        assert!(
            diff >= expected.saturating_sub(10) && diff <= expected + 10,
            "Timestamp delta should be ~4x OPERATION_TTL, got {diff}ms vs expected {expected}ms"
        );

        // multiplier(1) should be equivalent to ttl_transaction()
        let ttl_1x_via_multiplier = Transaction::ttl_transaction_with_multiplier(1);
        let diff_1x = ttl_1x
            .id
            .timestamp_ms()
            .abs_diff(ttl_1x_via_multiplier.id.timestamp_ms());
        assert!(
            diff_1x < 5,
            "multiplier(1) should be ~equivalent to ttl_transaction(), diff={diff_1x}ms"
        );
    }

    #[test]
    fn delta_or_full_state_delta_serialization_roundtrip() {
        use freenet_stdlib::prelude::StateDelta;

        let delta = StateDelta::from(vec![1, 2, 3, 4, 5]);
        let dofs = DeltaOrFullState::from_delta(&delta);

        // Serialize to bincode
        let serialized = bincode::serialize(&dofs).expect("serialize failed");

        // Deserialize back
        let deserialized: DeltaOrFullState =
            bincode::deserialize(&serialized).expect("deserialize failed");

        // Verify contents
        match &deserialized {
            DeltaOrFullState::Delta(bytes) => {
                assert_eq!(bytes, &vec![1, 2, 3, 4, 5]);
            }
            DeltaOrFullState::FullState(_) => panic!("expected Delta variant"),
        }

        // Verify to_delta works
        let recovered_delta = deserialized.to_delta().expect("should be delta");
        assert_eq!(recovered_delta.as_ref(), delta.as_ref());
    }

    #[test]
    fn delta_or_full_state_full_state_serialization_roundtrip() {
        use freenet_stdlib::prelude::State;

        let state = State::from(vec![10, 20, 30, 40, 50]);
        let dofs = DeltaOrFullState::from_state(&state);

        // Serialize to bincode
        let serialized = bincode::serialize(&dofs).expect("serialize failed");

        // Deserialize back
        let deserialized: DeltaOrFullState =
            bincode::deserialize(&serialized).expect("deserialize failed");

        // Verify contents
        match &deserialized {
            DeltaOrFullState::Delta(_) => panic!("expected FullState variant"),
            DeltaOrFullState::FullState(bytes) => {
                assert_eq!(bytes, &vec![10, 20, 30, 40, 50]);
            }
        }

        // Verify to_state works
        let recovered_state = deserialized.to_state().expect("should be full state");
        assert_eq!(recovered_state.as_ref(), state.as_ref());

        // Verify to_delta returns None for FullState
        assert!(deserialized.to_delta().is_none());
    }

    #[test]
    fn delta_or_full_state_conversion_methods() {
        use freenet_stdlib::prelude::{State, StateDelta};

        // Test from_delta
        let delta = StateDelta::from(vec![1, 2, 3]);
        let dofs = DeltaOrFullState::from_delta(&delta);
        assert!(matches!(dofs, DeltaOrFullState::Delta(_)));
        assert!(dofs.to_delta().is_some());
        assert!(dofs.to_state().is_none());

        // Test from_state
        let state = State::from(vec![4, 5, 6]);
        let dofs = DeltaOrFullState::from_state(&state);
        assert!(matches!(dofs, DeltaOrFullState::FullState(_)));
        assert!(dofs.to_delta().is_none());
        assert!(dofs.to_state().is_some());
    }

    #[test]
    fn delta_or_full_state_empty_data() {
        use freenet_stdlib::prelude::{State, StateDelta};

        // Empty delta
        let delta = StateDelta::from(Vec::<u8>::new());
        let dofs = DeltaOrFullState::from_delta(&delta);
        let serialized = bincode::serialize(&dofs).expect("serialize failed");
        let deserialized: DeltaOrFullState =
            bincode::deserialize(&serialized).expect("deserialize failed");
        assert!(matches!(deserialized, DeltaOrFullState::Delta(ref bytes) if bytes.is_empty()));

        // Empty state
        let state = State::from(Vec::<u8>::new());
        let dofs = DeltaOrFullState::from_state(&state);
        let serialized = bincode::serialize(&dofs).expect("serialize failed");
        let deserialized: DeltaOrFullState =
            bincode::deserialize(&serialized).expect("deserialize failed");
        assert!(matches!(deserialized, DeltaOrFullState::FullState(ref bytes) if bytes.is_empty()));
    }

    /// Verify SendInterestMessage Display produces compact output instead of
    /// dumping the full payload. This prevents regression to the 565KB-per-line
    /// log spam that caused 346MB/hr of gateway logs.
    #[test]
    fn test_send_interest_message_display_is_compact() {
        use std::net::SocketAddr;

        let addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();

        // Summaries with large payload should show count, not bytes
        let summaries = NodeEvent::SendInterestMessage {
            target: addr,
            message: InterestMessage::Summaries {
                entries: vec![
                    SummaryEntry {
                        hash: 123,
                        summary_bytes: Some(vec![0u8; 10_000]),
                    },
                    SummaryEntry {
                        hash: 456,
                        summary_bytes: Some(vec![0u8; 10_000]),
                    },
                ],
            },
        };
        let display = format!("{summaries}");
        assert!(
            display.len() < 200,
            "Display should be compact, got {} bytes: {display}",
            display.len()
        );
        assert!(
            display.contains("Summaries(2 entries)"),
            "Should show entry count: {display}"
        );

        // Interests should show hash count
        let interests = NodeEvent::SendInterestMessage {
            target: addr,
            message: InterestMessage::Interests {
                hashes: vec![1, 2, 3, 4, 5],
            },
        };
        let display = format!("{interests}");
        assert!(display.contains("Interests(5 hashes)"), "{display}");

        // ChangeInterests should show added/removed counts
        let changes = NodeEvent::SendInterestMessage {
            target: addr,
            message: InterestMessage::ChangeInterests {
                added: vec![1, 2],
                removed: vec![3],
            },
        };
        let display = format!("{changes}");
        assert!(
            display.contains("ChangeInterests(+2 -1 hashes)"),
            "{display}"
        );
    }
}