//! Codec wrapper for [Sender] and [Receiver].

use crate::{Blocker, CheckedSender, Receiver, Recipients, Sender};
use commonware_actor::{Feedback, Unreliable, mailbox};
use commonware_codec::{Codec, Decode, Encode, Error};
use commonware_cryptography::PublicKey;
use commonware_macros::select_loop;
use commonware_parallel::Strategy;
use commonware_runtime::{ContextCell, Handle, Metrics, Spawner, iobuf::EncodeExt, spawn_cell};
use commonware_utils::futures::Pool;
use std::{collections::VecDeque, time::SystemTime};

/// Wrap a [Sender] and [Receiver] with some [Codec].
pub const fn wrap<S: Sender, R: Receiver, V: Codec>(
    config: V::Cfg,
    sender: S,
    receiver: R,
) -> (WrappedSender<S, V>, WrappedReceiver<R, V>) {
    (
        WrappedSender::new(sender),
        WrappedReceiver::new(config, receiver),
    )
}

/// Tuple representing a message received from a given public key.
pub type WrappedMessage<P, V> = (P, Result<V, Error>);

/// Wrapper around a [Sender] that encodes messages using [Encode].
#[derive(Clone)]
pub struct WrappedSender<S: Sender, V: Encode> {
    sender: S,
    _phantom_v: std::marker::PhantomData<V>,
}

impl<S: Sender, V: Encode> WrappedSender<S, V> {
    /// Create a new [WrappedSender] with the given [Sender].
    pub const fn new(sender: S) -> Self {
        Self {
            sender,
            _phantom_v: std::marker::PhantomData,
        }
    }

    /// Send a message to a set of recipients.
    pub fn send(
        &mut self,
        recipients: Recipients<S::PublicKey>,
        message: V,
        priority: bool,
    ) -> Vec<S::PublicKey> {
        self.send_ref(recipients, &message, priority)
    }

    /// Send a borrowed message to a set of recipients.
    pub fn send_ref(
        &mut self,
        recipients: Recipients<S::PublicKey>,
        message: &V,
        priority: bool,
    ) -> Vec<S::PublicKey> {
        let encoded = message.encode_iobufs();
        self.sender.send(recipients, encoded, priority)
    }

    /// Check if a message can be sent to a set of recipients, returning a [CheckedWrappedSender]
    /// or the time at which the send can be retried.
    pub fn check(
        &mut self,
        recipients: Recipients<S::PublicKey>,
    ) -> Result<CheckedWrappedSender<'_, S, V>, SystemTime> {
        self.sender
            .check(recipients)
            .map(|checked| CheckedWrappedSender {
                sender: checked,
                _phantom_v: std::marker::PhantomData,
            })
    }
}

/// Checked sender that wraps a [`crate::LimitedSender::Checked`] and encodes messages using [Encode].
#[derive(Debug)]
pub struct CheckedWrappedSender<'a, S: Sender, V: Encode> {
    sender: S::Checked<'a>,
    _phantom_v: std::marker::PhantomData<V>,
}

impl<'a, S: Sender, V: Encode> CheckedWrappedSender<'a, S, V> {
    pub fn recipients(&self) -> Vec<S::PublicKey> {
        self.sender.recipients()
    }

    pub fn send(self, message: V, priority: bool) -> Unreliable<Feedback> {
        self.send_ref(&message, priority)
    }

    pub fn send_ref(self, message: &V, priority: bool) -> Unreliable<Feedback> {
        let encoded = message.encode_iobufs();
        self.sender.send(encoded, priority)
    }
}

/// Wrapper around a [Receiver] that decodes messages using [Decode].
pub struct WrappedReceiver<R: Receiver, V: Decode> {
    config: V::Cfg,
    receiver: R,
}

impl<R: Receiver, V: Decode> WrappedReceiver<R, V> {
    /// Create a new [WrappedReceiver] with the given [Receiver].
    pub const fn new(config: V::Cfg, receiver: R) -> Self {
        Self { config, receiver }
    }

    /// Receive a message from an arbitrary recipient.
    pub async fn recv(&mut self) -> Result<WrappedMessage<R::PublicKey, V>, R::Error> {
        let (pk, bytes) = self.receiver.recv().await?;
        let decoded = match V::decode_cfg(bytes, &self.config) {
            Ok(decoded) => decoded,
            Err(e) => {
                return Ok((pk, Err(e)));
            }
        };
        Ok((pk, Ok(decoded)))
    }
}

/// A decoded message with the peer that sent it.
struct Decoded<P: PublicKey, V>(P, V);

impl<P: PublicKey, V> mailbox::UnreliablePolicy for Decoded<P, V> {
    type Overflow = VecDeque<Self>;

    fn handle(_overflow: &mut Self::Overflow, _message: Self) -> bool {
        false
    }
}

/// Receiver half for successfully decoded messages from a [`WrappedBackgroundReceiver`].
pub struct BackgroundReceiver<P: PublicKey, V> {
    receiver: mailbox::UnreliableReceiver<Decoded<P, V>>,
}

impl<P: PublicKey, V> BackgroundReceiver<P, V> {
    /// Receive the next successfully decoded message.
    pub async fn recv(&mut self) -> Option<(P, V)> {
        self.receiver
            .recv()
            .await
            .map(|Decoded(peer, value)| (peer, value))
    }
}

/// A background receiver that receives raw bytes from a [`Receiver`] and spawns concurrent
/// decode tasks using [`Decode`].
///
/// Decode work is submitted to the provided [`Strategy`], so callers can offload expensive
/// decodes from the receive loop by choosing a parallel strategy. Each message is submitted as
/// soon as it is read, so the receive loop never waits on earlier decodes. The strategy's workers
/// bound how many decodes run at once.
///
/// Successfully decoded messages are forwarded through a mailbox that holds as many messages as
/// the wrapped [`Receiver`] reports through [`Receiver::capacity`]. If the consumer falls behind
/// and the mailbox fills, the receiver drops further decoded messages, since they would likely be
/// stale by the time the consumer reached them.
pub struct WrappedBackgroundReceiver<E, P, B, R, V, T>
where
    E: Spawner,
    P: PublicKey,
    B: Blocker<PublicKey = P>,
    R: Receiver<PublicKey = P>,
    V: Decode + Send,
    T: Strategy,
{
    context: ContextCell<E>,
    receiver: R,
    codec_config: V::Cfg,
    blocker: B,
    sender: mailbox::UnreliableSender<Decoded<P, V>>,
    strategy: T,
}

impl<E, P, B, R, V, T> WrappedBackgroundReceiver<E, P, B, R, V, T>
where
    E: Spawner + Metrics,
    P: PublicKey,
    B: Blocker<PublicKey = P>,
    R: Receiver<PublicKey = P>,
    V: Decode + Send + 'static,
    T: Strategy,
{
    /// Create a new [`WrappedBackgroundReceiver`].
    pub fn new(
        context: E,
        receiver: R,
        codec_config: V::Cfg,
        blocker: B,
        strategy: T,
    ) -> (Self, BackgroundReceiver<P, V>) {
        let (tx, rx) = mailbox::new_unreliable(context.child("mailbox"), receiver.capacity());
        (
            Self {
                context: ContextCell::new(context),
                receiver,
                codec_config,
                blocker,
                sender: tx,
                strategy,
            },
            BackgroundReceiver { receiver: rx },
        )
    }

    /// Start the background receiver.
    ///
    /// Returns a [`Handle`] that must be kept alive for the background receiver to continue
    /// running. Dropping the handle will abort the background receiver.
    pub fn start(mut self) -> Handle<()> {
        spawn_cell!(self.context, self.run())
    }

    /// Run the background receiver's event loop.
    ///
    /// Each incoming message is decoded through the provided strategy. A multi-worker strategy
    /// decodes on pool workers while the loop keeps draining the network buffer, and a strategy
    /// that runs jobs inline decodes on the receive loop.
    async fn run(mut self) {
        let mut decode_pool = Pool::default();
        let mut receiver_closed = false;

        select_loop! {
            self.context,
            on_start => {
                // Once the network receiver closes, forward the pending decodes and stop.
                if receiver_closed {
                    while !decode_pool.is_empty() {
                        let result = decode_pool.next_completed().await;
                        Self::handle_decode_result(&mut self.blocker, &mut self.sender, result);
                    }
                    break;
                }
            },
            on_stopped => {},
            // Process decode completions as they arrive
            result = decode_pool.next_completed() => {
                Self::handle_decode_result(&mut self.blocker, &mut self.sender, result);
            },
            // Receive raw bytes and submit decode work to the strategy.
            Ok((peer, bytes)) = self.receiver.recv() else {
                receiver_closed = true;
                continue;
            } => {
                let config = self.codec_config.clone();
                let handle = self.strategy.spawn(bytes.len(), move |_| {
                    let result = V::decode_cfg(bytes, &config);
                    (peer, result)
                });
                decode_pool.push(handle);
            },
        }
    }

    fn handle_decode_result(
        blocker: &mut B,
        sender: &mut mailbox::UnreliableSender<Decoded<P, V>>,
        result: (P, Result<V, commonware_codec::Error>),
    ) {
        let (peer, decode_result) = result;
        match decode_result {
            Ok(value) => {
                let _ = sender.enqueue(Decoded(peer, value));
            }
            Err(err) => {
                crate::block!(blocker, peer, ?err, "received invalid message");
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        Manager as _, Recipients,
        simulated::{self, Link, Network, Oracle},
    };
    use commonware_actor::Feedback;
    use commonware_codec::{Decode, Encode};
    use commonware_cryptography::{
        Signer,
        ed25519::{PrivateKey, PublicKey},
    };
    use commonware_macros::test_traced;
    use commonware_parallel::{Sequential, mocks};
    use commonware_runtime::{Clock as _, IoBuf, Quota, Runner, Supervisor as _, deterministic};
    use commonware_utils::{
        NZU32, NZUsize, Widen,
        channel::{mpsc, ring},
        ordered::Set,
        probability,
    };
    use std::{
        io,
        num::NonZeroUsize,
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
        time::Duration,
    };

    const LINK: Link = Link {
        latency: Duration::from_millis(0),
        jitter: Duration::from_millis(0),
        success_rate: probability!(1.0),
    };

    const TEST_QUOTA: Quota = Quota::per_second(NZU32!(64));

    fn start_network(context: deterministic::Context) -> Oracle<PublicKey, deterministic::Context> {
        let (network, oracle) = Network::new(
            context.child("network"),
            simulated::Config {
                max_size: 1024 * 1024,
                max_peers_per_set: NZUsize!(3),
                disconnect_on_block: true,
                tracked_peer_sets: NZUsize!(1),
            },
        );
        network.start();
        oracle
    }

    fn pk(seed: u64) -> PublicKey {
        PrivateKey::from_seed(seed).public_key()
    }

    fn track_peers<I>(oracle: &Oracle<PublicKey, deterministic::Context>, index: u64, peers: I)
    where
        I: IntoIterator<Item = PublicKey>,
    {
        oracle.manager().track(index, Set::from_iter_dedup(peers));
    }

    async fn link_bidirectional(
        oracle: &mut Oracle<PublicKey, deterministic::Context>,
        a: PublicKey,
        b: PublicKey,
    ) {
        oracle.add_link(a.clone(), b.clone(), LINK).await.unwrap();
        oracle.add_link(b, a, LINK).await.unwrap();
    }

    #[derive(Debug)]
    struct MockReceiver<P: commonware_cryptography::PublicKey> {
        receiver: mpsc::UnboundedReceiver<crate::Message<P>>,
        capacity: NonZeroUsize,
    }

    impl<P: commonware_cryptography::PublicKey> crate::Receiver for MockReceiver<P> {
        type Error = io::Error;
        type PublicKey = P;

        async fn recv(&mut self) -> Result<crate::Message<Self::PublicKey>, Self::Error> {
            self.receiver
                .recv()
                .await
                .ok_or_else(|| io::Error::from(io::ErrorKind::BrokenPipe))
        }

        fn capacity(&self) -> NonZeroUsize {
            self.capacity
        }
    }

    #[derive(Debug)]
    struct CountingReceiver<P: commonware_cryptography::PublicKey> {
        receiver: mpsc::UnboundedReceiver<crate::Message<P>>,
        received: Arc<AtomicUsize>,
        capacity: NonZeroUsize,
    }

    impl<P: commonware_cryptography::PublicKey> crate::Receiver for CountingReceiver<P> {
        type Error = io::Error;
        type PublicKey = P;

        async fn recv(&mut self) -> Result<crate::Message<Self::PublicKey>, Self::Error> {
            self.received.fetch_add(1, Ordering::SeqCst);
            self.receiver
                .recv()
                .await
                .ok_or_else(|| io::Error::from(io::ErrorKind::BrokenPipe))
        }

        fn capacity(&self) -> NonZeroUsize {
            self.capacity
        }
    }

    #[derive(Clone, Default)]
    struct NoopBlocker;

    impl crate::Blocker for NoopBlocker {
        type PublicKey = PublicKey;

        fn block(&mut self, _peer: Self::PublicKey) -> Feedback {
            Feedback::Ok
        }

        fn blocked(&mut self) -> crate::BlockedSubscription<Self::PublicKey> {
            let (_, receiver) = ring::channel(NZUsize!(1));
            receiver
        }
    }

    #[test_traced]
    fn test_valid_messages_forwarded() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut oracle = start_network(context.child("network"));

            let pk1 = pk(0);
            let pk2 = pk(1);
            let control1 = oracle.control(pk1.clone());
            let control2 = oracle.control(pk2.clone());
            track_peers(&oracle, 0, [pk1.clone(), pk2.clone()]);
            link_bidirectional(&mut oracle, pk1.clone(), pk2.clone()).await;

            let (mut sender1, _) = control1.register(0, TEST_QUOTA).await.unwrap();
            let (_, receiver2) = control2.register(0, TEST_QUOTA).await.unwrap();

            let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                receiver2,
                (),
                control2.clone(),
                Sequential,
            );
            let _handle = bg.start();

            let msg: u32 = 42;
            let _ = sender1.send(Recipients::One(pk2.clone()), msg.encode(), true);

            let (from, value) = rx.recv().await.unwrap();
            assert_eq!(from, pk1);
            assert_eq!(value, 42u32);
        });
    }

    #[test_traced]
    fn test_invalid_codec_blocks_peer() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut oracle = start_network(context.child("network"));

            let pk1 = pk(0);
            let pk2 = pk(1);
            let pk3 = pk(2);
            let control1 = oracle.control(pk1.clone());
            let control2 = oracle.control(pk2.clone());
            track_peers(&oracle, 0, [pk1.clone(), pk2.clone(), pk3.clone()]);
            link_bidirectional(&mut oracle, pk1.clone(), pk2.clone()).await;

            let (mut sender1, _) = control1.register(0, TEST_QUOTA).await.unwrap();
            let (_, receiver2) = control2.register(0, TEST_QUOTA).await.unwrap();

            let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                receiver2,
                (),
                control2.clone(),
                Sequential,
            );
            let _handle = bg.start();

            // Send a truncated payload (1 byte, but u32 needs 4).
            let invalid = IoBuf::from(vec![0xFFu8]);
            let _ = sender1.send(Recipients::One(pk2.clone()), invalid, true);

            // Then send a valid message from a different peer to confirm
            // the receiver is still running.
            let control3 = oracle.control(pk3.clone());
            link_bidirectional(&mut oracle, pk3.clone(), pk2.clone()).await;
            let (mut sender3, _) = control3.register(0, TEST_QUOTA).await.unwrap();

            let msg: u32 = 99;
            let _ = sender3.send(Recipients::One(pk2.clone()), msg.encode(), true);

            let (from, value) = rx.recv().await.unwrap();
            assert_eq!(from, pk3);
            assert_eq!(value, 99u32);

            // Verify pk1 was blocked.
            loop {
                let blocked = oracle.blocked().await.unwrap();
                if blocked.contains(&(pk2.clone(), pk1.clone())) {
                    break;
                }

                context.sleep(Duration::from_millis(1)).await;
            }
        });
    }

    #[test_traced]
    fn test_multiple_valid_messages() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut oracle = start_network(context.child("network"));

            let pk1 = pk(0);
            let pk2 = pk(1);
            let control1 = oracle.control(pk1.clone());
            let control2 = oracle.control(pk2.clone());
            track_peers(&oracle, 0, [pk1.clone(), pk2.clone()]);
            link_bidirectional(&mut oracle, pk1.clone(), pk2.clone()).await;

            let (mut sender1, _) = control1.register(0, TEST_QUOTA).await.unwrap();
            let (_, receiver2) = control2.register(0, TEST_QUOTA).await.unwrap();

            let count = 20;
            let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                receiver2,
                (),
                control2.clone(),
                Sequential,
            );
            let _handle = bg.start();

            for i in 0..count {
                let msg: u32 = i;
                let _ = sender1.send(Recipients::One(pk2.clone()), msg.encode(), true);
            }

            let mut received = Vec::new();
            for _ in 0..count {
                let (from, value) = rx.recv().await.unwrap();
                assert_eq!(from, pk1);
                received.push(value);
            }
            received.sort();
            assert_eq!(received, (0..count).collect::<Vec<u32>>());
        });
    }

    #[test_traced]
    fn test_decode_with_strategy() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut oracle = start_network(context.child("network"));

            let pk1 = pk(0);
            let pk2 = pk(1);
            let control1 = oracle.control(pk1.clone());
            let control2 = oracle.control(pk2.clone());
            track_peers(&oracle, 0, [pk1.clone(), pk2.clone()]);
            link_bidirectional(&mut oracle, pk1.clone(), pk2.clone()).await;

            let (mut sender1, _) = control1.register(0, TEST_QUOTA).await.unwrap();
            let (_, receiver2) = control2.register(0, TEST_QUOTA).await.unwrap();

            // Keep the message count below the mailbox capacity so every decode can be checked.
            let count = 50u32;
            let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                receiver2,
                (),
                control2.clone(),
                mocks::inline(NZUsize!(4)),
            );
            let _handle = bg.start();

            for i in 0..count {
                let _ = sender1.send(Recipients::One(pk2.clone()), i.encode(), true);
            }

            let mut received = Vec::new();
            for _ in 0..count {
                let (from, value) = rx.recv().await.unwrap();
                assert_eq!(from, pk1);
                received.push(value);
            }
            received.sort();
            assert_eq!(received, (0..count).collect::<Vec<u32>>());
        });
    }

    #[test_traced]
    fn test_invalid_among_valid_only_blocks_offender() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut oracle = start_network(context.child("network"));

            let pk1 = pk(0);
            let pk2 = pk(1);
            let pk3 = pk(2);
            let control1 = oracle.control(pk1.clone());
            let control2 = oracle.control(pk2.clone());
            let control3 = oracle.control(pk3.clone());
            track_peers(&oracle, 0, [pk1.clone(), pk2.clone(), pk3.clone()]);
            link_bidirectional(&mut oracle, pk1.clone(), pk2.clone()).await;
            link_bidirectional(&mut oracle, pk3.clone(), pk2.clone()).await;

            let (mut sender1, _) = control1.register(0, TEST_QUOTA).await.unwrap();
            let (_, receiver2) = control2.register(0, TEST_QUOTA).await.unwrap();
            let (mut sender3, _) = control3.register(0, TEST_QUOTA).await.unwrap();

            let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                receiver2,
                (),
                control2.clone(),
                Sequential,
            );
            let _handle = bg.start();

            // pk3 sends valid message.
            let _ = sender3.send(Recipients::One(pk2.clone()), 10u32.encode(), true);

            // pk1 sends invalid message.
            let _ = sender1.send(Recipients::One(pk2.clone()), IoBuf::from(vec![0xFF]), true);

            // pk3 sends another valid message.
            let _ = sender3.send(Recipients::One(pk2.clone()), 20u32.encode(), true);

            // Collect the two valid messages.
            let mut values = Vec::new();
            for _ in 0..2 {
                let (from, value) = rx.recv().await.unwrap();
                assert_eq!(from, pk3);
                values.push(value);
            }
            values.sort();
            assert_eq!(values, vec![10u32, 20]);

            // Only pk1 should be blocked.
            loop {
                let blocked = oracle.blocked().await.unwrap();
                assert!(!blocked.contains(&(pk2.clone(), pk3.clone())));
                if blocked.contains(&(pk2.clone(), pk1.clone())) {
                    break;
                }

                context.sleep(Duration::from_millis(1)).await;
            }
        });
    }

    /// The decoded mailbox retains exactly the wrapped receiver's capacity while the consumer
    /// is idle. Further decoded messages are dropped.
    #[test_traced]
    fn test_decoded_mailbox_holds_receiver_capacity() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // Queue more messages than the decoded mailbox can hold.
            let sender = pk(0);
            let (tx, receiver) = mpsc::unbounded_channel();

            let capacity = 3u32;
            for i in 0..capacity + 2 {
                tx.send((sender.clone(), IoBuf::from(i.encode())))
                    .expect("mock receiver should be open");
            }
            drop(tx);

            // Finish decoding before consuming any messages.
            let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                MockReceiver {
                    receiver,
                    capacity: NonZeroUsize::new(Widen::widen(capacity)).unwrap(),
                },
                (),
                NoopBlocker,
                Sequential,
            );
            let handle = bg.start();
            handle.await.expect("background receiver should complete");

            // Only the first `capacity` messages survive.
            for expected in 0..capacity {
                let (from, value) = rx.recv().await.unwrap();
                assert_eq!(from, sender);
                assert_eq!(value, expected);
            }
            assert!(rx.recv().await.is_none());
        });
    }

    /// The receiver reads every message while none of their decodes complete.
    #[test_traced]
    fn test_pending_decodes_do_not_stall_raw_receives() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let sender = pk(0);
            let (tx, receiver) = mpsc::unbounded_channel();
            let received = Arc::new(AtomicUsize::new(0));

            let count = 10u32;
            for i in 0..count {
                tx.send((sender.clone(), IoBuf::from(i.encode())))
                    .expect("mock receiver should be open");
            }

            let (bg, _rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                CountingReceiver {
                    receiver,
                    received: received.clone(),
                    capacity: NZUsize!(16),
                },
                (),
                NoopBlocker,
                mocks::pending(NZUsize!(2)),
            );
            let handle = bg.start();

            // The receiver reads all `count` messages while their decodes hang on the pool, then
            // calls `recv` once more.
            let reads = Widen::widen(count + 1);
            while received.load(Ordering::SeqCst) < reads {
                context.sleep(Duration::from_millis(1)).await;
            }

            drop(handle);
        });
    }

    #[test_traced]
    fn test_drain_decode_pool_after_receiver_closure() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let sender = pk(0);
            let (tx, receiver) = mpsc::unbounded_channel();
            let count = 64u32;

            for i in 0..count {
                tx.send((sender.clone(), IoBuf::from(i.encode())))
                    .expect("mock receiver should be open");
            }
            drop(tx);

            let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, u32, _>::new(
                context.child("bg"),
                MockReceiver {
                    receiver,
                    capacity: NZUsize!(64),
                },
                (),
                NoopBlocker,
                Sequential,
            );
            let _handle = bg.start();

            let mut values = Vec::new();
            while let Some((from, value)) = rx.recv().await {
                assert_eq!(from, sender);
                values.push(value);
            }
            values.sort_unstable();

            assert_eq!(values, (0..count).collect::<Vec<u32>>());
        });
    }

    #[test_traced]
    fn test_recv_view() {
        let executor = deterministic::Runner::default();
        executor.start(|_| async move {
            let sender = pk(0);
            let value: Vec<IoBuf> = (0..8).map(|_| IoBuf::from(vec![1u8; 17])).collect();
            let frame = value.encode();
            let cfg = ((..).into(), (..).into());
            let range = frame.as_ptr_range();

            // Decoding the received frame by value hands out views of it
            let (tx, rx) = mpsc::unbounded_channel();
            tx.send((sender.clone(), IoBuf::from(frame.clone())))
                .expect("mock receiver should be open");
            let mut receiver = WrappedReceiver::<_, Vec<IoBuf>>::new(
                cfg,
                MockReceiver {
                    receiver: rx,
                    capacity: NZUsize!(1),
                },
            );
            let (from, decoded) = receiver.recv().await.unwrap();
            let decoded = decoded.unwrap();
            assert_eq!(from, sender);
            assert_eq!(decoded, value);
            assert!(decoded.iter().all(|b| range.contains(&b.as_ref().as_ptr())));

            // Decoding a slice of the frame copies every field
            let copied = Vec::<IoBuf>::decode_cfg(commonware_codec::Copying(&frame), &cfg).unwrap();
            assert_eq!(copied, value);
            assert!(copied.iter().all(|b| !range.contains(&b.as_ref().as_ptr())));
        });
    }

    #[test_traced]
    fn test_background_recv_view() {
        deterministic::Runner::default().start(|context| async move {
            let sender = pk(0);
            let value: Vec<IoBuf> = (0..8).map(|_| IoBuf::from(vec![1u8; 17])).collect();
            let encoded = value.encode();
            let frames = [
                ("external", IoBuf::from(encoded.clone())),
                ("native", IoBuf::copy_from_slice(&encoded)),
            ];
            drop(encoded);

            for (label, frame) in frames {
                let range = frame.as_ref().as_ptr_range();
                let (tx, receiver) = mpsc::unbounded_channel();
                tx.send((sender.clone(), frame.clone())).unwrap();
                drop(tx);
                let (bg, mut rx) = WrappedBackgroundReceiver::<_, _, _, _, Vec<IoBuf>, _>::new(
                    context.child(label),
                    MockReceiver {
                        receiver,
                        capacity: NZUsize!(1),
                    },
                    ((..).into(), (..).into()),
                    NoopBlocker,
                    mocks::inline(NZUsize!(2)),
                );
                bg.start().await.unwrap();

                let (from, decoded) = rx.recv().await.unwrap();
                assert_eq!(from, sender);
                assert!(
                    decoded
                        .iter()
                        .all(|field| range.contains(&field.as_ref().as_ptr()))
                );
                drop(frame);
                assert_eq!(decoded, value);
                assert!(rx.recv().await.is_none());
            }
        });
    }
}