pipecircuit 0.0.20

use core::marker::PhantomData;

/// A nominal duplex semantic contract, independent of its implementation and transport.
/// Submission does not imply a synchronous response; either direction may be uninhabited.
pub trait Pipe {
    type In<'message>;
    type Out<'message>;
}

pub enum NoInput {}

pub enum NoOutput {}

/// Moves one value into a particular pipe boundary.
///
/// The destination pipe is part of the contract so an implementation
/// can keep distinct destinations nominal even when their inputs are equal.
/// Outcomes remain precise: inline transport completes a step now, while a
/// queued transport may instead report admission, backpressure, or failure.
pub trait TransportTo<Destination: Pipe> {
    type Outcome<'transport, 'message>
    where
        Self: 'transport;

    #[must_use = "transport outcomes must be routed or settled"]
    fn send<'transport, 'message>(
        &'transport mut self,
        input: Destination::In<'message>,
    ) -> Self::Outcome<'transport, 'message>;
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[must_use = "received outputs, quiescence, backpressure, and failures must be handled"]
pub enum ReceiveFromOutcome<Output, Rejected, Failure> {
    Received { output: Output },
    Progressed,
    Quiescent,
    Backpressured { rejected: Rejected },
    Failed { failure: Failure },
}

impl<Output, Rejected, Failure> ReceiveFromOutcome<Output, Rejected, Failure> {
    /// Adapt rejected ownership without changing progress, output, or failure.
    pub fn map_rejected<Mapped>(
        self,
        map: impl FnOnce(Rejected) -> Mapped,
    ) -> ReceiveFromOutcome<Output, Mapped, Failure> {
        match self {
            Self::Received { output } => ReceiveFromOutcome::Received { output },
            Self::Progressed => ReceiveFromOutcome::Progressed,
            Self::Quiescent => ReceiveFromOutcome::Quiescent,
            Self::Backpressured { rejected } => ReceiveFromOutcome::Backpressured {
                rejected: map(rejected),
            },
            Self::Failed { failure } => ReceiveFromOutcome::Failed { failure },
        }
    }

    /// Adapt the exact failure without changing output or rejected ownership.
    pub fn map_failure<Mapped>(
        self,
        map: impl FnOnce(Failure) -> Mapped,
    ) -> ReceiveFromOutcome<Output, Rejected, Mapped> {
        match self {
            Self::Received { output } => ReceiveFromOutcome::Received { output },
            Self::Progressed => ReceiveFromOutcome::Progressed,
            Self::Quiescent => ReceiveFromOutcome::Quiescent,
            Self::Backpressured { rejected } => ReceiveFromOutcome::Backpressured { rejected },
            Self::Failed { failure } => ReceiveFromOutcome::Failed {
                failure: map(failure),
            },
        }
    }
}

pub trait TransportFrom<Source: Pipe> {
    type IncomingMessage<'message>
    where
        Self: 'message;
    type Failure;

    fn receive_from<'message>(
        &'message mut self,
        incoming_message: Self::IncomingMessage<'message>,
    ) -> ReceiveFromOutcome<Source::Out<'message>, Self::IncomingMessage<'message>, Self::Failure>;
}

pub trait DuplexTransport<Protocol: Pipe> {
    type Incoming<'transport>: TransportFrom<Protocol>
    where
        Self: 'transport;
    type Outgoing<'transport>: TransportTo<Protocol>
    where
        Self: 'transport;

    fn split<'transport>(
        &'transport mut self,
    ) -> (Self::Incoming<'transport>, Self::Outgoing<'transport>);
}

/// A lifetime-indexed family of values that may be encoded or decoded.
pub trait MessageFamily {
    type Message<'message>;
}

/// Selects the message family accepted by an input pipe.
pub struct PipeIn<Selected: Pipe>(PhantomData<fn() -> Selected>);

impl<Selected: Pipe> MessageFamily for PipeIn<Selected> {
    type Message<'message> = Selected::In<'message>;
}

pub struct PipeOut<Selected: Pipe>(PhantomData<fn() -> Selected>);

impl<Selected: Pipe> MessageFamily for PipeOut<Selected> {
    type Message<'message> = Selected::Out<'message>;
}

/// Selects one owned message type as a lifetime-indexed family.
pub struct OwnedMessage<Message>(PhantomData<fn() -> Message>);

impl<Message> MessageFamily for OwnedMessage<Message> {
    type Message<'message> = Message;
}

/// The byte count written by a message encoder.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct EncodedByteCount(usize);

impl EncodedByteCount {
    pub const fn new(value: usize) -> Self {
        Self(value)
    }

    pub const fn get(self) -> usize {
        self.0
    }
}

/// Proves that one lifetime-indexed message family can be serialized into
/// bounded caller-owned storage.
pub trait MessageEncoder {
    type Message<'message>: ?Sized;
    type Error;

    fn encode<'message>(
        &self,
        message: &Self::Message<'message>,
        destination: &mut [u8],
    ) -> Result<EncodedByteCount, Self::Error>;
}

/// Proves that one serialized message can become a typed value.
///
/// The decoded value may borrow from the reusable receive frame.
pub trait MessageDecoder {
    type Message<'frame>;
    type Error;

    fn decode<'frame>(&self, frame: &'frame mut [u8])
    -> Result<Self::Message<'frame>, Self::Error>;
}

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

    struct Pipe;

    impl super::Pipe for Pipe {
        type In<'message> = &'message str;
        type Out<'message> = &'message [u8];
    }

    struct Source;

    impl super::Pipe for Source {
        type In<'message> = NoInput;
        type Out<'message> = &'message str;
    }

    struct Sink;

    impl super::Pipe for Sink {
        type In<'message> = &'message str;
        type Out<'message> = NoOutput;
    }

    struct BorrowedInline;

    impl TransportTo<Pipe> for BorrowedInline {
        type Outcome<'transport, 'message>
            = &'message str
        where
            Self: 'transport;

        fn send<'transport, 'message>(
            &'transport mut self,
            input: <Pipe as super::Pipe>::In<'message>,
        ) -> Self::Outcome<'transport, 'message> {
            input
        }
    }

    #[test]
    fn participant_message_families_include_uninhabited_directions() {
        fn assert_message_families<Selected>()
        where
            Selected: super::Pipe,
            for<'message> PipeIn<Selected>:
                MessageFamily<Message<'message> = Selected::In<'message>>,
            for<'message> PipeOut<Selected>:
                MessageFamily<Message<'message> = Selected::Out<'message>>,
        {
        }

        assert_message_families::<Pipe>();
        assert_message_families::<Source>();
        assert_message_families::<Sink>();
    }

    #[test]
    fn encoded_byte_counts_preserve_their_value() {
        assert_eq!(EncodedByteCount::new(7).get(), 7);
    }

    #[test]
    fn outcome_adapters_preserve_every_variant_and_only_map_the_selected_payload() {
        use core::cell::Cell;
        type Outcome = ReceiveFromOutcome<u8, u16, u32>;
        for (outcome, expected, should_map) in [
            (
                Outcome::Received { output: 11 },
                Outcome::Received { output: 11 },
                false,
            ),
            (Outcome::Progressed, Outcome::Progressed, false),
            (Outcome::Quiescent, Outcome::Quiescent, false),
            (
                Outcome::Backpressured { rejected: 17 },
                Outcome::Backpressured { rejected: 43 },
                true,
            ),
            (
                Outcome::Failed { failure: 23 },
                Outcome::Failed { failure: 23 },
                false,
            ),
        ] {
            let mapped = Cell::new(false);
            assert_eq!(
                outcome.map_rejected(|rejected| {
                    assert_eq!(rejected, 17);
                    mapped.set(true);
                    43
                }),
                expected
            );
            assert_eq!(mapped.get(), should_map);
        }
        for (outcome, expected, should_map) in [
            (
                Outcome::Received { output: 11 },
                Outcome::Received { output: 11 },
                false,
            ),
            (Outcome::Progressed, Outcome::Progressed, false),
            (Outcome::Quiescent, Outcome::Quiescent, false),
            (
                Outcome::Backpressured { rejected: 17 },
                Outcome::Backpressured { rejected: 17 },
                false,
            ),
            (
                Outcome::Failed { failure: 23 },
                Outcome::Failed { failure: 37 },
                true,
            ),
        ] {
            let mapped = Cell::new(false);
            assert_eq!(
                outcome.map_failure(|failure| {
                    assert_eq!(failure, 23);
                    mapped.set(true);
                    37
                }),
                expected
            );
            assert_eq!(mapped.get(), should_map);
        }
    }

    #[test]
    fn failure_mapping_preserves_exclusive_ownership() -> Result<(), &'static str> {
        let mut bytes = [3, 7];
        let address = bytes.as_ptr();
        let outcome = ReceiveFromOutcome::<(), (), _>::Failed {
            failure: &mut bytes,
        }
        .map_failure(|failure| ("worker", failure));
        let ReceiveFromOutcome::Failed {
            failure: ("worker", failure),
        } = outcome
        else {
            return Err("failure ownership changed");
        };
        assert_eq!(failure.as_ptr(), address);
        assert_eq!(failure, &[3, 7]);
        failure.fill(9);
        assert_eq!(bytes, [9, 9]);
        Ok(())
    }

    #[test]
    fn transports_without_internal_steps_preserve_borrowed_delivery() {
        let mut transport = BorrowedInline;
        assert_eq!(
            TransportTo::<Pipe>::send(&mut transport, "borrowed"),
            "borrowed"
        );
    }

    #[test]
    fn outcome_adapters_preserve_exclusive_borrows_without_calling_unrelated_mappers()
    -> Result<(), &'static str> {
        use core::cell::Cell;
        let mut value = 3_u8;
        let mapped = Cell::new(false);
        let received = ReceiveFromOutcome::<_, (), ()>::Received { output: &mut value }
            .map_rejected(|()| {
                mapped.set(true);
                1_u8
            })
            .map_failure(|()| {
                mapped.set(true);
                2_u8
            });
        let ReceiveFromOutcome::Received { output } = received else {
            return Err("received ownership changed");
        };
        *output = 4;
        assert_eq!(value, 4);
        assert!(!mapped.get());

        let rejected = ReceiveFromOutcome::<(), _, ()>::Backpressured {
            rejected: &mut value,
        }
        .map_failure(|()| {
            mapped.set(true);
            2_u8
        })
        .map_rejected(|rejected| ("source", rejected));
        let ReceiveFromOutcome::Backpressured {
            rejected: ("source", rejected),
        } = rejected
        else {
            return Err("rejected ownership changed");
        };
        *rejected = 5;
        assert_eq!(value, 5);
        assert!(!mapped.get());
        Ok(())
    }
}