pipecircuit 0.0.23

use core::marker::PhantomData;

use crate::{Pipe, StateMachine, StepInputOf};

/// 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>);
}

mod private {
    use super::{Pipe, StateMachine, StepInputOf};

    pub(super) trait StepLaw<'message, Protocol: Pipe> {}

    impl<'message, Machine, Protocol> StepLaw<'message, Protocol> for Machine
    where
        Machine: StateMachine,
        Protocol: Pipe,
        Protocol::In<'message>: StepInputOf<Machine, Outcome = Protocol::Out<'message>>,
    {
    }
}

#[doc(hidden)]
#[allow(private_bounds)]
pub trait StepsPipeInputToOutput<'message, Protocol: Pipe>:
    StateMachine + private::StepLaw<'message, Protocol>
{
    #[must_use = "pipe step outcomes must be routed or settled"]
    #[doc(hidden)]
    fn step_boundary(&mut self, input: Protocol::In<'message>) -> Protocol::Out<'message>;
}

impl<'message, Machine, Protocol> StepsPipeInputToOutput<'message, Protocol> for Machine
where
    Machine: StateMachine,
    Protocol: Pipe,
    Protocol::In<'message>: StepInputOf<Machine, Outcome = Protocol::Out<'message>>,
{
    fn step_boundary(&mut self, input: Protocol::In<'message>) -> Protocol::Out<'message> {
        input.step(self)
    }
}

/// A synchronous transport bound to one named pipe and one local implementation.
/// Use `for_pipe` when the protocol is separate from the implementing machine.
pub struct Inline<Implementation, Protocol = Implementation> {
    destination: Implementation,
    protocol: PhantomData<fn() -> Protocol>,
}

impl<Implementation> Inline<Implementation> {
    pub const fn new(destination: Implementation) -> Self {
        Self {
            destination,
            protocol: PhantomData,
        }
    }

    pub const fn for_pipe<Protocol: Pipe>(
        destination: Implementation,
    ) -> Inline<Implementation, Protocol> {
        Inline {
            destination,
            protocol: PhantomData,
        }
    }
}

impl<Implementation, Protocol> Inline<Implementation, Protocol> {
    pub const fn destination(&self) -> &Implementation {
        &self.destination
    }

    pub fn into_destination(self) -> Implementation {
        self.destination
    }
}

impl<Implementation, Protocol> TransportTo<Protocol> for Inline<Implementation, Protocol>
where
    Protocol: Pipe,
    Implementation: for<'message> StepsPipeInputToOutput<'message, Protocol>,
{
    type Outcome<'transport, 'message>
        = Protocol::Out<'message>
    where
        Self: 'transport;

    fn send<'transport, 'message>(
        &'transport mut self,
        input: Protocol::In<'message>,
    ) -> Self::Outcome<'transport, 'message> {
        self.destination.step_boundary(input)
    }
}

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

    struct Borrowed;

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

    struct BorrowedInline;

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

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

    struct Counter(u8);

    impl StateMachine for Counter {}

    struct Add(u8);

    impl Pipe for Counter {
        type In<'message> = Add;
        type Out<'message> = u8;
    }

    impl StepInputOf<Counter> for Add {
        type Outcome = u8;

        fn step(self, counter: &mut Counter) -> Self::Outcome {
            counter.0 = counter.0.saturating_add(self.0);
            counter.0
        }
    }

    #[test]
    fn inline_transport_completes_the_precise_step() {
        let mut transport = Inline::new(Counter(3));
        assert_eq!(TransportTo::<Counter>::send(&mut transport, Add(4)), 7);
        assert_eq!(transport.into_destination().0, 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::<Borrowed>::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(())
    }
}