dope-manifold 0.26.0

use std::{ops::ControlFlow, pin};

use dope_core::driver::{
    retained, route,
    route::Identity,
    schedule::{self, ready},
};
use dope_net::{
    link::{
        event,
        pool::recv::{self, Readiness, Ready, arrival, delivery, reserve::Choice},
    },
    wire,
};

use crate::app::receive::{Mode, ingress};

#[derive(Clone, Copy)]
pub(crate) enum Replay {
    First,
    Waiting,
}

pub(crate) trait Admission<'a, 'c: 'a, 'owner: 'a, 'd: 'a + 'c + 'owner>:
    Sized
    + Into<(
        &'a mut retained::Context<'c, 'owner, 'd>,
        schedule::Turn<'a, 'd>,
    )>
{
    fn fresh(
        turn: schedule::Turn<'a, 'd>,
        driver: &'a mut retained::Context<'c, 'owner, 'd>,
    ) -> schedule::Capacity<Self>;

    fn replay(
        replay: Replay,
        ready: ready::Key<'d>,
        turn: schedule::Turn<'a, 'd>,
        driver: &'a mut retained::Context<'c, 'owner, 'd>,
    ) -> schedule::Capacity<Self>;
}

pub(crate) trait Strategy<W: wire::Spec, M: Mode<W>> {
    type Value<'input, 'w, 'd>
    where
        'd: 'w,
        'w: 'input,
        W: 'w + 'd;

    type Pending<'w, 'd>
    where
        'd: 'w,
        W: 'w + 'd;

    type Admission<'a, 'c, 'owner, 'd>: Admission<'a, 'c, 'owner, 'd>
    where
        'c: 'a,
        'owner: 'a,
        'd: 'a + 'c + 'owner,
        W: 'd;

    fn drive_data<'p, 'a, 'c, 'owner, 'w, 'd, ID: Identity + 'd, P, A>(
        policy: pin::Pin<&mut P>,
        data: A,
        work: Self::Admission<'a, 'c, 'owner, 'd>,
    ) where
        'c: 'a,
        'owner: 'a,
        'd: 'a + 'c + 'owner + 'w + 'p,
        W: 'w + 'd,
        P: ingress::Policy<'p, 'w, 'd, ID, Wire = W, Input = M>,
        M::Reserve: Choice<W, M::Ingress>,
        A: arrival::Arrival<'d, Observed = event::Data<'d>>;

    fn resume_tail<'p, 'w, 'd, ID: Identity + 'd, P>(
        _: pin::Pin<&mut P>,
        _: &delivery::Resumed<'d, ID>,
        _: bool,
        _: schedule::Turn<'_, 'd>,
        _: &mut retained::Context<'_, '_, 'd>,
    ) -> ControlFlow<(), bool>
    where
        'd: 'w + 'p,
        W: 'w + 'd,
        P: ingress::Policy<'p, 'w, 'd, ID, Wire = W, Input = M>,
        M::Reserve: Choice<W, M::Ingress>,
    {
        ControlFlow::Continue(false)
    }

    fn resume<'p, 'w, 'd, ID: Identity + 'd, P>(
        mut policy: pin::Pin<&mut P>,
        target: route::Target<'d, ID>,
        turn: schedule::Turn<'_, 'd>,
        driver: &mut retained::Context<'_, '_, 'd>,
    ) where
        'd: 'w + 'p,
        W: 'w + 'd,
        P: ingress::Policy<'p, 'w, 'd, ID, Wire = W, Input = M>,
        M::Reserve: Choice<W, M::Ingress>,
    {
        let item = match P::storage(policy.as_mut()).recv().resume(target) {
            recv::Resumption::Ready(item) => item,
            recv::Resumption::Unavailable(_) => return,
        };
        let ready = item.ready();
        let mut first = true;
        loop {
            if let ControlFlow::Break(wake) =
                P::admit(policy.as_mut(), || Some(target), turn.reborrow(), driver)
            {
                wake.wake();
                break;
            }
            match Self::resume_tail(policy.as_mut(), &item, first, turn.reborrow(), driver) {
                ControlFlow::Break(()) => break,
                ControlFlow::Continue(true) => {
                    first = false;
                    continue;
                }
                ControlFlow::Continue(false) => {}
            }
            let key = item.key();
            let (end, done) = {
                let inspect = P::storage(policy.as_mut()).inspect();
                let done = matches!(inspect.recv(key), Readiness::Ready);
                let end = inspect.get(key).is_some_and(|slot| {
                    slot.terminal.has_eof() || (!done && slot.terminal.kind().is_some())
                });
                (end, done)
            };
            if end {
                ingress::prepare_terminal(policy.as_mut(), key);
                match P::terminal(policy.as_mut(), key, turn.reborrow(), driver) {
                    ingress::Terminal::Ended(end) => {
                        ingress::finish_eof(policy.as_mut(), key, end, turn.reborrow(), driver);
                        first = false;
                        continue;
                    }
                    ingress::Terminal::Pending | ingress::Terminal::Closed => break,
                    ingress::Terminal::Idle => {}
                }
            }
            if !done {
                break;
            }
            let replay = if first {
                Replay::First
            } else {
                Replay::Waiting
            };
            let work = match Self::Admission::replay(replay, ready, turn.reborrow(), driver) {
                schedule::Capacity::Available(work) => work,
                schedule::Capacity::Exhausted => break,
            };
            let done = match P::storage(policy.as_mut()).recv().pop_ready(&item) {
                Ready::Completion(done) => done,
                Ready::Empty | Ready::Paused | Ready::Stale => break,
            };
            first = false;
            match done.classify() {
                event::Kind::Data(data) => Self::drive_data(policy.as_mut(), data, work),
                event::Kind::Control(ctrl) => {
                    let (driver, step) = work.into();
                    ingress::drive_control(policy.as_mut(), ctrl, step, driver);
                }
            }
        }
        P::flush(policy, turn, driver);
    }
}