use crate::{Ingress, authenticated::data::Data};
use commonware_codec::{
    Buf, Encode, EncodeSize, Error as CodecError, FixedSize, MaxEncodeSize, Read, ReadExt, Write,
    config::RangeCfg,
    varint::{MAX_U32_VARINT_SIZE, MAX_U64_VARINT_SIZE, UInt},
};
use commonware_cryptography::PublicKey;
use commonware_runtime::{BufMut, Clock};
use commonware_utils::SystemTimeExt;
use std::time::Duration;
use thiserror::Error;

/// Errors that can occur when interacting with [crate::authenticated::discovery::types].
#[derive(Error, Debug)]
pub enum Error {
    #[error("too many peers: {0}")]
    TooManyPeers(usize),
    #[error("received self")]
    ReceivedSelf,
    #[error("invalid signature")]
    InvalidSignature,
    #[error("synchrony bound violated")]
    SynchronyBound,
}

/// Prefix byte used to identify a [Payload] with variant Data.
const DATA_PREFIX: u8 = crate::authenticated::data::DATA_PREFIX; // 0
/// Prefix byte used to identify a [Payload] with variant Greeting.
const GREETING_PREFIX: u8 = 1;
/// Prefix byte used to identify a [Payload] with variant BitVec.
const BIT_VEC_PREFIX: u8 = 2;
/// Prefix byte used to identify a [Payload] with variant Peers.
const PEERS_PREFIX: u8 = 3;

// Use chunk size of 1 to minimize encoded size.
type BitMap = commonware_utils::bitmap::BitMap<1>;

/// Configuration for deserializing [Payload].
///
/// This is used to limit the size of the messages received from peers.
#[derive(Clone)]
pub struct PayloadConfig {
    /// The maximum number of bits that can be sent in a `BitVec` message.
    pub max_bit_vec: u64,

    /// The maximum number of peers that can be sent in a `Peers` message.
    pub max_peers: usize,

    /// The maximum length of the data that can be sent in a `Data` message.
    pub max_data_length: usize,
}

/// Payload is the only allowed message format that can be sent between peers.
#[derive(Clone, Debug)]
pub enum Payload<C: PublicKey> {
    /// Arbitrary data sent between peers.
    Data(Data),

    /// A greeting message containing the peer's own information.
    ///
    /// This must be the first message sent after connection establishment.
    /// The connection will be terminated if this message is not received first
    /// or if it is received more than once.
    Greeting(Info<C>),

    /// Bit vector that represents the peers a peer knows about.
    ///
    /// Also used as a ping message to keep the connection alive.
    BitVec(BitVec),

    /// A vector of verifiable peer information.
    Peers(Vec<Info<C>>),
}

/// Returns an upper bound on the encoded size of any [Payload::Greeting], [Payload::BitVec], or
/// [Payload::Peers] that [Payload::read_cfg] admits under `max_bit_vec` and `max_peers`.
///
/// Gossip shares its stream with application data, so the stream's frame limit must admit this
/// size.
pub fn max_gossip_size<C: PublicKey>(max_bit_vec: u64, max_peers: usize) -> usize {
    let info = Info::<C>::max_encode_size();
    let bit_vec = MAX_U64_VARINT_SIZE + BitMap::max_encode_size(max_bit_vec);
    let peers = max_peers * info + MAX_U32_VARINT_SIZE;
    u8::SIZE + info.max(bit_vec).max(peers)
}

impl<C: PublicKey> EncodeSize for Payload<C> {
    fn encode_size(&self) -> usize {
        (match self {
            Self::Data(data) => data.encode_size(),
            Self::Greeting(info) => info.encode_size(),
            Self::BitVec(bit_vec) => bit_vec.encode_size(),
            Self::Peers(peers) => peers.encode_size(),
        }) + 1
    }
}

impl<C: PublicKey> Write for Payload<C> {
    fn write(&self, buf: &mut impl BufMut) {
        match self {
            Self::Data(data) => {
                DATA_PREFIX.write(buf);
                data.write(buf);
            }
            Self::Greeting(info) => {
                GREETING_PREFIX.write(buf);
                info.write(buf);
            }
            Self::BitVec(bit_vec) => {
                BIT_VEC_PREFIX.write(buf);
                bit_vec.write(buf);
            }
            Self::Peers(peers) => {
                PEERS_PREFIX.write(buf);
                peers.write(buf);
            }
        }
    }
}

impl<C: PublicKey> Read for Payload<C> {
    type Cfg = PayloadConfig;

    fn read_cfg(buf: &mut impl Buf, cfg: &Self::Cfg) -> Result<Self, CodecError> {
        let PayloadConfig {
            max_bit_vec,
            max_peers,
            max_data_length,
        } = cfg;

        let payload_type = <u8>::read(buf)?;
        match payload_type {
            DATA_PREFIX => {
                let data = Data::read_cfg(buf, &(..=*max_data_length).into())?;
                Ok(Self::Data(data))
            }
            GREETING_PREFIX => {
                let info = Info::<C>::read(buf)?;
                Ok(Self::Greeting(info))
            }
            BIT_VEC_PREFIX => {
                let bit_vec = BitVec::read_cfg(buf, max_bit_vec)?;
                Ok(Self::BitVec(bit_vec))
            }
            PEERS_PREFIX => {
                let peers = Vec::<Info<C>>::read_cfg(buf, &(RangeCfg::new(..=*max_peers), ()))?;
                Ok(Self::Peers(peers))
            }
            other => Err(CodecError::InvalidEnum(other)),
        }
    }
}

#[cfg(feature = "arbitrary")]
impl<C: PublicKey> arbitrary::Arbitrary<'_> for Payload<C>
where
    C: for<'a> arbitrary::Arbitrary<'a>,
    C::Signature: for<'a> arbitrary::Arbitrary<'a>,
{
    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
        let choice = u.int_in_range(0..=3)?;
        match choice {
            0 => Ok(Self::Data(u.arbitrary()?)),
            1 => Ok(Self::Greeting(u.arbitrary()?)),
            2 => Ok(Self::BitVec(u.arbitrary()?)),
            3 => Ok(Self::Peers(u.arbitrary()?)),
            _ => unreachable!(),
        }
    }
}

/// BitVec is a bit vector that represents the peers a peer knows about at a given index.
///
/// A peer should respond with a `Peers` message if they know of any peers that the sender does not.
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
pub struct BitVec {
    /// The index that the bit vector applies to.
    pub index: u64,

    /// The bit vector itself.
    pub bits: BitMap,
}

impl EncodeSize for BitVec {
    fn encode_size(&self) -> usize {
        UInt(self.index).encode_size() + self.bits.encode_size()
    }
}

impl Write for BitVec {
    fn write(&self, buf: &mut impl BufMut) {
        UInt(self.index).write(buf);
        self.bits.write(buf);
    }
}

impl Read for BitVec {
    type Cfg = u64;

    fn read_cfg(buf: &mut impl Buf, max_bits: &u64) -> Result<Self, CodecError> {
        let index = UInt::read(buf)?.into();
        let bits = BitMap::read_cfg(buf, &(..=*max_bits).into())?;
        Ok(Self { index, bits })
    }
}

/// A signed message from a peer attesting to its own ingress address and public key at a given time.
///
/// This is used to share the peer's ingress address and public key with other peers in a verified
/// manner.
#[derive(Clone, Debug)]
pub struct Info<C: PublicKey> {
    /// The ingress address of the peer (how to dial them).
    pub ingress: Ingress,

    /// The timestamp (epoch milliseconds) at which the ingress was signed over.
    pub timestamp: u64,

    /// The public key of the peer.
    pub public_key: C,

    /// The peer's signature over the ingress and timestamp.
    pub signature: C::Signature,
}

impl<C: PublicKey> Info<C> {
    /// Returns an upper bound on the encoded size of any [Info].
    pub const fn max_encode_size() -> usize {
        Ingress::MAX_ENCODE_SIZE + MAX_U64_VARINT_SIZE + C::SIZE + C::Signature::SIZE
    }

    /// Verify the signature of [Info].
    pub fn verify(&self, namespace: &[u8]) -> bool {
        self.public_key.verify(
            namespace,
            &(self.ingress.clone(), self.timestamp).encode(),
            &self.signature,
        )
    }

    /// Create a new [InfoVerifier] with the provided configuration.
    pub const fn verifier(
        me: C,
        peer_gossip_max_count: usize,
        synchrony_bound: Duration,
        ip_namespace: Vec<u8>,
    ) -> InfoVerifier<C> {
        InfoVerifier::new(me, peer_gossip_max_count, synchrony_bound, ip_namespace)
    }

    /// Sign the [Info] message with the supplied identity and signing function.
    ///
    /// The signing function must produce signatures verifiable by `public_key`.
    pub fn sign(
        public_key: C,
        namespace: &[u8],
        ingress: impl Into<Ingress>,
        timestamp: u64,
        sign: impl FnOnce(&[u8], &[u8]) -> C::Signature,
    ) -> Self {
        let ingress = ingress.into();
        let signature = sign(namespace, &(ingress.clone(), timestamp).encode());
        Self {
            ingress,
            timestamp,
            public_key,
            signature,
        }
    }
}

impl<C: PublicKey> EncodeSize for Info<C> {
    fn encode_size(&self) -> usize {
        self.ingress.encode_size()
            + UInt(self.timestamp).encode_size()
            + self.public_key.encode_size()
            + self.signature.encode_size()
    }
}

impl<C: PublicKey> Write for Info<C> {
    fn write(&self, buf: &mut impl BufMut) {
        self.ingress.write(buf);
        UInt(self.timestamp).write(buf);
        self.public_key.write(buf);
        self.signature.write(buf);
    }
}

impl<C: PublicKey> Read for Info<C> {
    type Cfg = ();

    fn read_cfg(buf: &mut impl Buf, _cfg: &Self::Cfg) -> Result<Self, CodecError> {
        let ingress = Ingress::read(buf)?;
        let timestamp = UInt::read(buf)?.into();
        let public_key = C::read(buf)?;
        let signature = C::Signature::read(buf)?;
        Ok(Self {
            ingress,
            timestamp,
            public_key,
            signature,
        })
    }
}

#[cfg(feature = "arbitrary")]
impl<C: PublicKey> arbitrary::Arbitrary<'_> for Info<C>
where
    C: for<'a> arbitrary::Arbitrary<'a>,
    C::Signature: for<'a> arbitrary::Arbitrary<'a>,
{
    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
        let ingress = u.arbitrary()?;
        let timestamp = u.arbitrary()?;
        let public_key = u.arbitrary()?;
        let signature = u.arbitrary()?;

        Ok(Self {
            ingress,
            timestamp,
            public_key,
            signature,
        })
    }
}

/// Validate peer gossip payloads against configurability and basic safety checks.
#[derive(Clone)]
pub struct InfoVerifier<C: PublicKey> {
    /// The [PublicKey] of the verifier.
    me: C,

    /// The maximum number of [Info] allowable in a single message.
    peer_gossip_max_count: usize,

    /// The time bound for synchrony. Messages with timestamps greater than this far into the
    /// future will be considered malformed.
    synchrony_bound: Duration,

    /// The namespace used to sign and verify [Info] messages.
    ip_namespace: Vec<u8>,
}

impl<C: PublicKey> InfoVerifier<C> {
    /// Create a new [InfoVerifier] with the provided configuration.
    const fn new(
        me: C,
        peer_gossip_max_count: usize,
        synchrony_bound: Duration,
        ip_namespace: Vec<u8>,
    ) -> Self {
        Self {
            me,
            peer_gossip_max_count,
            synchrony_bound,
            ip_namespace,
        }
    }

    /// Validate a list of peer information from a gossip message.
    ///
    /// Returns an error if the list itself or any entries can be considered malformed.
    pub fn validate(&self, clock: &impl Clock, infos: &[Info<C>]) -> Result<(), Error> {
        // Ensure there aren't too many peers sent
        if infos.len() > self.peer_gossip_max_count {
            return Err(Error::TooManyPeers(infos.len()));
        }

        // We allow peers to be sent in any order when responding to a bit vector (allows
        // for selecting a random subset of peers when there are too many) and allow
        // for duplicates (no need to create an additional set to check this)
        infos
            .iter()
            .try_for_each(|info| self.validate_info(clock, info))
    }

    /// Validate one peer's self-signed information, such as a connection greeting.
    ///
    /// Returns an error if the information can be considered malformed.
    pub fn validate_info(&self, clock: &impl Clock, info: &Info<C>) -> Result<(), Error> {
        // Check if peer is us
        if info.public_key == self.me {
            return Err(Error::ReceivedSelf);
        }

        // Check if timestamp is too far into the future
        if Duration::from_millis(info.timestamp)
            > clock.current().epoch().saturating_add(self.synchrony_bound)
        {
            return Err(Error::SynchronyBound);
        }

        // Check if signature is valid
        if !info.verify(self.ip_namespace.as_ref()) {
            return Err(Error::InvalidSignature);
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::authenticated::{
        MAX_PAYLOAD_OVERHEAD,
        data::allocations::{allocation_calls_during, frame},
    };
    use commonware_codec::{Decode, DecodeExt};
    use commonware_cryptography::{
        Signer,
        secp256r1::standard::{PrivateKey, PublicKey},
    };
    use commonware_math::algebra::Random;
    use commonware_runtime::{Clock, IoBuf, Runner, deterministic};
    use commonware_utils::{hostname, test_rng};
    use std::{net::SocketAddr, time::Duration};

    const NAMESPACE: &[u8] = b"test";

    fn signed_peer_info(rng: &mut impl rand_core::CryptoRng) -> Info<PublicKey> {
        let signer = PrivateKey::random(rng);
        Info {
            ingress: Ingress::Socket(SocketAddr::from(([127, 0, 0, 1], 8080))),
            timestamp: 1234567890,
            public_key: signer.public_key(),
            signature: signer.sign(NAMESPACE, &[1, 2, 3, 4, 5]),
        }
    }

    #[test]
    fn test_data_prefix_value() {
        assert_eq!(DATA_PREFIX, 0);
    }

    #[test]
    fn test_bit_vec_codec() {
        let original = BitVec {
            index: 1234,
            bits: BitMap::ones(71),
        };
        let decoded = BitVec::decode_cfg(original.encode(), &71).unwrap();
        assert_eq!(original, decoded);
    }

    #[test]
    fn test_signed_peer_info_codec() {
        let mut rng = test_rng();
        let original = vec![
            signed_peer_info(&mut rng),
            signed_peer_info(&mut rng),
            signed_peer_info(&mut rng),
        ];
        let encoded = original.encode();
        let decoded =
            Vec::<Info<PublicKey>>::decode_cfg(encoded, &(RangeCfg::new(3..=3), ())).unwrap();
        for (original, decoded) in original.iter().zip(decoded.iter()) {
            assert_eq!(original.ingress, decoded.ingress);
            assert_eq!(original.timestamp, decoded.timestamp);
            assert_eq!(original.public_key, decoded.public_key);
            assert_eq!(original.signature, decoded.signature);
        }

        let too_short =
            Vec::<Info<PublicKey>>::decode_cfg(original.encode(), &(RangeCfg::new(..3), ()));
        assert!(matches!(too_short, Err(CodecError::InvalidLength(3))));

        let too_long =
            Vec::<Info<PublicKey>>::decode_cfg(original.encode(), &(RangeCfg::new(4..), ()));
        assert!(matches!(too_long, Err(CodecError::InvalidLength(3))));
    }

    #[test]
    fn test_payload_codec() {
        let mut rng = test_rng();

        // Config for the codec
        let cfg = PayloadConfig {
            max_bit_vec: 1024,
            max_peers: 10,
            max_data_length: 100,
        };

        // Test Greeting
        let original = signed_peer_info(&mut rng);
        let encoded = Payload::Greeting(original.clone()).encode();
        let decoded = match Payload::<PublicKey>::decode_cfg(encoded, &cfg) {
            Ok(Payload::<PublicKey>::Greeting(info)) => info,
            _ => panic!("Expected Greeting payload"),
        };
        assert_eq!(original.ingress, decoded.ingress);
        assert_eq!(original.timestamp, decoded.timestamp);
        assert_eq!(original.public_key, decoded.public_key);
        assert_eq!(original.signature, decoded.signature);

        // Test BitVec
        let original = BitVec {
            index: 1234,
            bits: BitMap::ones(100),
        };
        let encoded = Payload::<PublicKey>::BitVec(original.clone()).encode();
        let decoded = match Payload::<PublicKey>::decode_cfg(encoded, &cfg) {
            Ok(Payload::<PublicKey>::BitVec(b)) => b,
            _ => panic!(),
        };
        assert_eq!(original, decoded);

        // Test Peers
        let original = vec![signed_peer_info(&mut rng), signed_peer_info(&mut rng)];
        let encoded = Payload::Peers(original.clone()).encode();
        let decoded = match Payload::<PublicKey>::decode_cfg(encoded, &cfg) {
            Ok(Payload::<PublicKey>::Peers(p)) => p,
            _ => panic!(),
        };
        for (a, b) in original.iter().zip(decoded.iter()) {
            assert_eq!(a.ingress, b.ingress);
            assert_eq!(a.timestamp, b.timestamp);
            assert_eq!(a.public_key, b.public_key);
            assert_eq!(a.signature, b.signature);
        }

        // Test Data
        let original = Data {
            channel: 12345,
            message: IoBuf::from(b"Hello, world!"),
        };
        let encoded = Payload::<PublicKey>::Data(original.clone()).encode();
        let decoded = match Payload::<PublicKey>::decode_cfg(encoded, &cfg) {
            Ok(Payload::<PublicKey>::Data(d)) => d,
            _ => panic!(),
        };
        assert_eq!(original, decoded);
    }

    #[test]
    fn test_payload_decode_invalid_type() {
        let cfg = PayloadConfig {
            max_bit_vec: 1024,
            max_peers: 10,
            max_data_length: 100,
        };
        // Type byte 4 is invalid (Data=0, Greeting=1, BitVec=2, Peers=3)
        let invalid_payload = [4, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
        let result =
            Payload::<PublicKey>::decode_cfg(commonware_codec::Copying(&invalid_payload), &cfg);
        assert!(result.is_err());
    }

    #[test]
    fn test_payload_bitvec_respects_limit() {
        let cfg = PayloadConfig {
            max_bit_vec: 8,
            max_peers: 10,
            max_data_length: 32,
        };
        let encoded = Payload::<PublicKey>::BitVec(BitVec {
            index: 5,
            bits: BitMap::ones(9),
        })
        .encode();
        let err = Payload::<PublicKey>::decode_cfg(encoded, &cfg).unwrap_err();
        assert!(matches!(err, CodecError::InvalidLength(9)));
    }

    #[test]
    fn test_payload_peers_respects_limit() {
        let mut rng = test_rng();
        let cfg = PayloadConfig {
            max_bit_vec: 1024,
            max_peers: 1,
            max_data_length: 32,
        };
        let peers = vec![signed_peer_info(&mut rng), signed_peer_info(&mut rng)];
        let encoded = Payload::Peers(peers).encode();
        let err = Payload::<PublicKey>::decode_cfg(encoded, &cfg).unwrap_err();
        assert!(matches!(err, CodecError::InvalidLength(2)));
    }

    #[test]
    fn test_payload_data_respects_limit() {
        let cfg = PayloadConfig {
            max_bit_vec: 1024,
            max_peers: 10,
            max_data_length: 4,
        };
        let encoded = Payload::<PublicKey>::Data(Data {
            channel: 1,
            message: IoBuf::from(b"hello"),
        })
        .encode();
        let err = Payload::<PublicKey>::decode_cfg(encoded, &cfg).unwrap_err();
        assert!(matches!(err, CodecError::InvalidLength(5)));
    }

    #[test]
    fn test_max_payload_data_overhead() {
        let message = IoBuf::from(vec![0; 1 << 29]);
        let message_len = message.len();
        let payload = Payload::<PublicKey>::Data(Data {
            channel: u64::MAX,
            message,
        });
        assert_eq!(
            payload.encode_size(),
            message_len + MAX_PAYLOAD_OVERHEAD as usize
        );
    }

    #[test]
    fn test_max_gossip_size_bounds_payloads() {
        let mut rng = test_rng();
        let signer = PrivateKey::random(&mut rng);

        // A maximum-length hostname and timestamp give the largest possible info. Its bound also
        // reserves the hostname length prefix at its type's maximum width.
        let host = [
            "a".repeat(63),
            "b".repeat(63),
            "c".repeat(63),
            "d".repeat(61),
        ]
        .join(".");
        let info = Info::sign(
            signer.public_key(),
            NAMESPACE,
            Ingress::Dns {
                host: hostname!(host),
                port: u16::MAX,
            },
            u64::MAX,
            |namespace, message| signer.sign(namespace, message),
        );
        let ingress_slack = Ingress::MAX_ENCODE_SIZE - info.ingress.encode_size();
        assert_eq!(
            info.encode_size() + ingress_slack,
            Info::<PublicKey>::max_encode_size()
        );

        let (max_bit_vec, max_peers) = (1_000, 32);
        let bound = max_gossip_size::<PublicKey>(max_bit_vec, max_peers);
        let greeting = Payload::Greeting(info.clone());
        let bit_vec = Payload::<PublicKey>::BitVec(BitVec {
            index: u64::MAX,
            bits: BitMap::ones(max_bit_vec),
        });
        let peers = Payload::Peers(vec![info; max_peers]);
        assert!(greeting.encode_size() <= bound);
        assert!(bit_vec.encode_size() <= bound);
        assert!(peers.encode_size() <= bound);

        // The peer-list bound reserves each length prefix at its type's maximum width.
        let count_slack = MAX_U32_VARINT_SIZE - max_peers.encode_size();
        assert_eq!(
            bound,
            peers.encode_size() + count_slack + max_peers * ingress_slack
        );

        // Without peer gossip, the largest bit vector determines the bound exactly.
        let max_bit_vec = 100_000;
        let bit_vec = Payload::<PublicKey>::BitVec(BitVec {
            index: u64::MAX,
            bits: BitMap::ones(max_bit_vec),
        });
        assert_eq!(
            max_gossip_size::<PublicKey>(max_bit_vec, 0),
            bit_vec.encode_size()
        );
    }

    #[test]
    fn info_verifier_accepts_valid_peer() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let validator_signer = PrivateKey::random(&mut context);
            let signer = PrivateKey::random(&mut context);
            let validator = Info::verifier(
                validator_signer.public_key(),
                4,
                Duration::from_secs(30),
                NAMESPACE.to_vec(),
            );
            let timestamp = context.current().epoch().as_millis() as u64;
            let peer = Info::sign(
                signer.public_key(),
                NAMESPACE,
                SocketAddr::from(([8, 8, 8, 8], 8080)),
                timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            assert!(validator.validate(&context, &[peer]).is_ok());
        });
    }

    #[test]
    fn info_verifier_rejects_too_many_peers() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let signer = PrivateKey::random(&mut context);
            let synchrony_bound = Duration::from_secs(30);
            let timestamp = context.current().epoch().as_millis() as u64;
            let peers = {
                let addr_a = SocketAddr::from(([8, 8, 8, 8], 9000));
                let addr_b = SocketAddr::from(([8, 8, 4, 4], 9001));
                let signer_a = PrivateKey::random(&mut context);
                let peer_a = Info::sign(
                    signer_a.public_key(),
                    NAMESPACE,
                    addr_a,
                    timestamp,
                    |namespace, message| signer_a.sign(namespace, message),
                );
                let signer_b = PrivateKey::random(&mut context);
                let peer_b = Info::sign(
                    signer_b.public_key(),
                    NAMESPACE,
                    addr_b,
                    timestamp,
                    |namespace, message| signer_b.sign(namespace, message),
                );
                vec![peer_a, peer_b]
            };
            let validator =
                Info::verifier(signer.public_key(), 1, synchrony_bound, NAMESPACE.to_vec());
            let err = validator.validate(&context, &peers).unwrap_err();
            assert!(matches!(err, Error::TooManyPeers(count) if count == 2));
        });
    }

    #[test]
    fn info_verifier_rejects_self() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let signer = PrivateKey::random(&mut context);
            let validator = Info::verifier(
                signer.public_key(),
                4,
                Duration::from_secs(30),
                NAMESPACE.to_vec(),
            );
            let timestamp = context.current().epoch().as_millis() as u64;
            let peer = Info::sign(
                signer.public_key(),
                NAMESPACE,
                SocketAddr::from(([203, 0, 113, 1], 8080)),
                timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            let err = validator.validate(&context, &[peer]).unwrap_err();
            assert!(matches!(err, Error::ReceivedSelf));
        });
    }

    #[test]
    fn info_verifier_rejects_future_timestamp() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let validator_signer = PrivateKey::random(&mut context);
            let signer = PrivateKey::random(&mut context);
            let synchrony_bound = Duration::from_secs(30);
            let validator = Info::verifier(
                validator_signer.public_key(),
                4,
                synchrony_bound,
                NAMESPACE.to_vec(),
            );
            let future_timestamp =
                (context.current().epoch() + synchrony_bound + Duration::from_secs(1)).as_millis()
                    as u64;
            let peer = Info::sign(
                signer.public_key(),
                NAMESPACE,
                SocketAddr::from(([198, 51, 100, 1], 8080)),
                future_timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            let err = validator.validate(&context, &[peer]).unwrap_err();
            assert!(matches!(err, Error::SynchronyBound));
        });
    }

    #[test]
    fn info_verifier_allows_past_timestamp() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let validator_signer = PrivateKey::random(&mut context);
            let signer = PrivateKey::random(&mut context);
            let synchrony_bound = Duration::from_secs(30);
            let validator = Info::verifier(
                validator_signer.public_key(),
                4,
                synchrony_bound,
                NAMESPACE.to_vec(),
            );

            // Advance current time
            context.sleep(synchrony_bound * 2).await;

            // Create a peer whose timestamp is older than the synchrony bound allows
            let past_timestamp =
                (context.current().epoch() - synchrony_bound - Duration::from_secs(1)).as_millis()
                    as u64;
            let peer = Info::sign(
                signer.public_key(),
                NAMESPACE,
                SocketAddr::from(([198, 51, 100, 1], 8080)),
                past_timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            assert!(validator.validate(&context, &[peer]).is_ok());
        });
    }

    #[test]
    fn info_verifier_rejects_invalid_signature() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let validator_signer = PrivateKey::random(&mut context);
            let signer = PrivateKey::random(&mut context);
            let validator = Info::verifier(
                validator_signer.public_key(),
                4,
                Duration::from_secs(30),
                NAMESPACE.to_vec(),
            );
            let timestamp = context.current().epoch().as_millis() as u64;
            let peer = Info::sign(
                signer.public_key(),
                b"wrong-namespace",
                SocketAddr::from(([8, 8, 4, 4], 8080)),
                timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            let err = validator.validate(&context, &[peer]).unwrap_err();
            assert!(matches!(err, Error::InvalidSignature));
        });
    }

    #[test]
    fn info_with_dns_ingress_sign_and_verify() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let signer = PrivateKey::random(&mut context);
            let timestamp = context.current().epoch().as_millis() as u64;
            let dns_ingress = Ingress::Dns {
                host: hostname!("node.example.com"),
                port: 8080,
            };
            let peer = Info::sign(
                signer.public_key(),
                NAMESPACE,
                dns_ingress.clone(),
                timestamp,
                |namespace, message| signer.sign(namespace, message),
            );

            assert_eq!(peer.ingress, dns_ingress);
            assert_eq!(peer.timestamp, timestamp);
            assert_eq!(peer.public_key, signer.public_key());
            assert!(peer.verify(NAMESPACE));
        });
    }

    #[test]
    fn info_with_dns_ingress_codec() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let signer = PrivateKey::random(&mut context);
            let timestamp = context.current().epoch().as_millis() as u64;
            let dns_ingress = Ingress::Dns {
                host: hostname!("validator-1.network.io"),
                port: 9090,
            };
            let original = Info::sign(
                signer.public_key(),
                NAMESPACE,
                dns_ingress.clone(),
                timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            let encoded = original.encode();
            let decoded = Info::<PublicKey>::decode(encoded).unwrap();

            assert_eq!(decoded.ingress, dns_ingress);
            assert_eq!(decoded.timestamp, timestamp);
            assert_eq!(decoded.public_key, original.public_key);
            assert_eq!(decoded.signature, original.signature);
            assert!(decoded.verify(NAMESPACE));
        });
    }

    #[test]
    fn info_verifier_accepts_dns_ingress() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let validator_signer = PrivateKey::random(&mut context);
            let signer = PrivateKey::random(&mut context);
            let validator = Info::verifier(
                validator_signer.public_key(),
                4,
                Duration::from_secs(30),
                NAMESPACE.to_vec(),
            );
            let timestamp = context.current().epoch().as_millis() as u64;
            let dns_ingress = Ingress::Dns {
                host: hostname!("peer.network.com"),
                port: 8080,
            };
            let peer = Info::sign(
                signer.public_key(),
                NAMESPACE,
                dns_ingress,
                timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            assert!(validator.validate(&context, &[peer]).is_ok());
        });
    }

    #[test]
    fn info_verifier_accepts_dns_ingress_with_internal_hostname() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let validator_signer = PrivateKey::random(&mut context);
            let signer = PrivateKey::random(&mut context);
            let validator = Info::verifier(
                validator_signer.public_key(),
                4,
                Duration::from_secs(30),
                NAMESPACE.to_vec(),
            );
            let timestamp = context.current().epoch().as_millis() as u64;
            let dns_ingress = Ingress::Dns {
                host: hostname!("internal.local"),
                port: 8080,
            };
            let peer = Info::sign(
                signer.public_key(),
                NAMESPACE,
                dns_ingress,
                timestamp,
                |namespace, message| signer.sign(namespace, message),
            );
            assert!(
                validator.validate(&context, &[peer]).is_ok(),
                "DNS ingress should be accepted"
            );
        });
    }

    #[test]
    fn info_verifier_validates_greeting_without_gossip_limit() {
        let executor = deterministic::Runner::default();
        executor.start(|mut context| async move {
            let validator_signer = PrivateKey::random(&mut context);
            let signer = PrivateKey::random(&mut context);

            // A gossip limit of zero disables peer gossip, but every connection still starts
            // with one self-signed greeting.
            let validator = Info::verifier(
                validator_signer.public_key(),
                0,
                Duration::from_secs(30),
                NAMESPACE.to_vec(),
            );
            let greeting = Info::sign(
                signer.public_key(),
                NAMESPACE,
                SocketAddr::from(([8, 8, 8, 8], 8080)),
                context.current().epoch_millis(),
                |namespace, message| signer.sign(namespace, message),
            );
            assert!(validator.validate_info(&context, &greeting).is_ok());
            assert!(matches!(
                validator.validate(&context, &[greeting]),
                Err(Error::TooManyPeers(1))
            ));
        });
    }

    #[cfg(feature = "arbitrary")]
    mod conformance {
        use super::*;
        use commonware_codec::conformance::CodecConformance;

        commonware_conformance::conformance_tests! {
            CodecConformance<Payload<PublicKey>> => 1024,
            CodecConformance<BitVec>,
            CodecConformance<Info<PublicKey>> => 1024,
        }
    }

    /// Decoding a data frame through the payload type the peer actor uses neither copies the
    /// message nor allocates owner records for it.
    #[commonware_macros::test_group("miri")]
    #[test]
    fn test_decode_does_not_allocate() {
        let original = Data {
            channel: 12345,
            message: IoBuf::from(b"Hello, world!"),
        };
        let (frame, bytes) = frame(&Payload::<PublicKey>::Data(original.clone()).encode());
        let cfg = PayloadConfig {
            max_bit_vec: 8,
            max_peers: 8,
            max_data_length: 1024,
        };
        let (decoded, calls) =
            allocation_calls_during(|| Payload::<PublicKey>::decode_cfg(frame, &cfg).unwrap());
        assert_eq!(calls, 0);
        let Payload::Data(decoded) = decoded else {
            panic!("expected a data frame");
        };
        assert_eq!(decoded, original);
        assert!(bytes.contains(&decoded.message.as_ptr()));
    }
}