pipecircuit 0.0.5

use core::num::NonZeroU64;
use core::{cmp::Ordering, hash::Hash as _};

extern crate std;

use self::std::{collections::hash_map::DefaultHasher, format};

use super::*;
use crate::{Participant, Transport};

struct Machine {
    settled: Option<(NonZeroU64, u8)>,
}

impl StateMachine for Machine {}

struct Completion(u8);
struct Expiry(u8);
struct Work(u8);

struct Worker {
    handled: Option<u8>,
}

impl StateMachine for Worker {}

impl StepInputOf<Worker> for Work {
    type Outcome = Completion;

    fn step(self, worker: &mut Worker) -> Self::Outcome {
        worker.handled = Some(self.0);
        Completion(self.0)
    }
}

impl Participant for Worker {
    type In<'message> = ContinuationWork<Machine, Work>;
    type Out<'message> = ContinuationCompletion<Machine, Completion>;
}

impl StepInputOf<Machine> for ContinuationInput<Machine, Completion, Expiry> {
    type Outcome = (NonZeroU64, u8);

    fn step(self, machine: &mut Machine) -> Self::Outcome {
        let (id, settlement) = self.into_parts();
        let value = match settlement {
            ContinuationSettlement::Completed(Completion(value))
            | ContinuationSettlement::Expired(Expiry(value)) => value,
        };
        let settled = (id.value(), value);
        machine.settled = Some(settled);
        settled
    }
}

fn timeout_after() -> PositiveDuration {
    PositiveDuration::from_nanoseconds(NonZeroU64::new(17).unwrap_or(NonZeroU64::MIN))
}

fn owed(work: u8, expiry: u8) -> Owed<Machine, Completion, Expiry, Work> {
    Owed::new(Work(work), Expiry(expiry), timeout_after())
}

fn issue(
    coordinator: &mut ContinuationCoordinator<Machine>,
    work: u8,
    expiry: u8,
) -> Result<UnparkedOwedParts<Machine, Completion, Expiry, Work>, ()> {
    coordinator.try_issue(owed(work, expiry)).map_err(|_| ())
}

#[test]
fn identifiers_are_monotonic_nonzero_and_never_wrap() -> Result<(), ()> {
    let mut ids = ContinuationIds::<Machine>::new();
    assert_eq!(ids.try_issue().map_err(|_| ())?.value().get(), 1);
    assert_eq!(ids.try_issue().map_err(|_| ())?.value().get(), 2);

    ids.next = NextContinuationId::Available(NonZeroU64::MAX);
    assert_eq!(ids.try_issue().map_err(|_| ())?.value(), NonZeroU64::MAX);
    assert_eq!(ids.try_issue(), Err(ContinuationIdentifiersExhausted));
    Ok(())
}

#[test]
fn semantic_debt_exists_before_identity_and_survives_exhaustion() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator {
        ids: ContinuationIds {
            next: NextContinuationId::Exhausted,
            machine: PhantomData,
        },
    };
    let owed = owed(7, 11);
    assert_eq!(owed.work().0, 7);

    let failure = coordinator.try_issue(owed).err().ok_or(())?;
    let recovered = failure.into_owed();
    assert_eq!(recovered.work().0, 7);
    assert_eq!(recovered.timeout_after(), timeout_after());
    Ok(())
}

#[test]
fn issued_work_deadline_and_authority_share_one_identity() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let owed = owed(7, 11);
    assert_eq!(owed.work().0, 7);
    assert_eq!(owed.timeout_after(), timeout_after());

    let (work, deadline, continuation) = coordinator.try_issue(owed).map_err(|_| ())?.into_parts();
    let id = work.id();
    assert_eq!(work.work().0, 7);
    assert_eq!(deadline.id(), id);
    assert_eq!(deadline.after(), timeout_after());
    assert_eq!(continuation.id(), id);
    Ok(())
}

#[test]
fn wait_is_the_unit_work_specialization() {
    let owed = Owed::<Machine, Completion, Expiry>::wait(Expiry(11), timeout_after());
    assert_eq!(owed.work(), &());
    assert_eq!(owed.timeout_after(), timeout_after());
}

#[test]
fn completion_routes_one_input_and_one_deadline_cancellation() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let (_, _, continuation) = issue(&mut coordinator, 7, 11)?.into_parts();
    let id = continuation.id();
    let (input, cancellation) = continuation.complete(Completion(13)).into_parts();

    assert_eq!(input.id(), id);
    assert_eq!(cancellation.id(), id);
    assert!(matches!(
        input.settlement(),
        ContinuationSettlement::Completed(Completion(13))
    ));
    Ok(())
}

#[test]
fn expiration_routes_the_exact_stored_expiry_input() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let (_, _, continuation) = issue(&mut coordinator, 7, 11)?.into_parts();
    let id = continuation.id();
    let input = continuation.expire();

    assert_eq!(input.id(), id);
    assert!(matches!(
        input.settlement(),
        ContinuationSettlement::Expired(Expiry(11))
    ));
    Ok(())
}

#[test]
fn coordinator_parks_before_releasing_the_deadline_and_work() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::default();
    let mut slot = ContinuationSlot::new();
    let parked = coordinator
        .try_admit(owed(7, 11), &mut slot)
        .map_err(|_| ())?;
    let id = slot.pending_id().ok_or(())?;
    let (deadline, work) = parked.into_parts();
    assert_eq!(deadline.id(), id);
    assert_eq!(work.id(), id);
    assert_eq!(work.work().0, 7);

    let completed = coordinator
        .try_complete(&mut slot, ContinuationCompletion::new(id, Completion(13)))
        .map_err(|_| ())?;
    let (input, cancellation) = completed.into_parts();
    assert_eq!(cancellation.id(), id);
    assert!(matches!(
        input.settlement(),
        ContinuationSettlement::Completed(Completion(13))
    ));
    assert!(matches!(
        coordinator.try_expire(&mut slot, id),
        Err(ContinuationNotPending::Vacant { requested }) if requested == id
    ));
    Ok(())
}

#[test]
fn addressed_work_steps_the_target_and_returns_an_addressed_completion() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let mut slot = ContinuationSlot::new();
    let parked = coordinator
        .try_admit(owed(7, 11), &mut slot)
        .map_err(|_| ())?;
    let (deadline, work) = parked.into_parts();
    let mut worker = crate::Inline::new(Worker { handled: None });

    let completion = Transport::<Worker>::transport(
        &mut worker,
        work,
        &mut crate::TurnContext::new(crate::NoopTurnObserver),
    );

    assert_eq!(worker.destination().handled, Some(7));
    assert_eq!(completion.id(), deadline.id());
    assert_eq!(completion.completion().0, 7);
    let mapped = completion.map(|Completion(value)| value == 7);
    assert!(mapped.completion());
    Ok(())
}

#[test]
fn decoded_correlation_rebuilds_work_but_not_settlement_authority() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let mut slot = ContinuationSlot::new();
    let admitted = coordinator
        .try_admit(owed(7, 11), &mut slot)
        .map_err(|_| ())?;
    let (deadline, work) = admitted.into_parts();
    let (id, work) = work.into_parts();

    let decoded_id = ContinuationId::from_nonzero(id.value());
    let decoded_work = ContinuationWork::new(decoded_id, work);
    let mut target = Worker { handled: None };
    let completion = target.step(decoded_work);
    let completed = coordinator
        .try_complete(&mut slot, completion)
        .map_err(|_| ())?;

    assert_eq!(completed.into_parts().1.id(), deadline.id());
    assert_eq!(target.handled, Some(7));
    Ok(())
}

#[test]
fn expiry_wins_the_opposite_ordering_and_retains_machine_routing() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let mut slot = ContinuationSlot::new();
    let parked = coordinator
        .try_admit(owed(7, 11), &mut slot)
        .map_err(|_| ())?;
    let (deadline, _work) = parked.into_parts();
    let id = deadline.id();

    let input = coordinator.try_expire(&mut slot, id).map_err(|_| ())?;
    let mut machine = Machine { settled: None };
    assert_eq!(machine.step(input), (id.value(), 11));
    assert_eq!(machine.settled, Some((id.value(), 11)));

    let failure = coordinator
        .try_complete(&mut slot, ContinuationCompletion::new(id, Completion(13)))
        .err()
        .ok_or(())?;
    assert!(matches!(
        failure.cause(),
        ContinuationNotPending::Vacant { requested } if *requested == id
    ));
    assert_eq!(failure.into_completion().completion().0, 13);
    Ok(())
}

#[test]
fn a_different_identity_cannot_displace_the_pending_authority() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let mut slot = ContinuationSlot::new();
    let (_, _, first) = issue(&mut coordinator, 1, 2)?.into_parts();
    let first_id = first.id();
    let (_, _, second) = issue(&mut coordinator, 3, 4)?.into_parts();
    let second_id = second.id();
    slot.try_park(first).map_err(|_| ())?;

    assert!(matches!(
        slot.try_take(second_id),
        Err(ContinuationNotPending::Different { requested, pending })
            if requested == second_id && pending == first_id
    ));
    assert_eq!(slot.pending_id(), Some(first_id));
    assert_eq!(slot.try_take(first_id).map_err(|_| ())?.id(), first_id);
    Ok(())
}

#[test]
fn occupied_slots_return_the_rejected_authority_untouched() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let mut slot = ContinuationSlot::new();
    let (_, _, first) = issue(&mut coordinator, 1, 2)?.into_parts();
    let first_id = first.id();
    let (_, _, second) = issue(&mut coordinator, 3, 4)?.into_parts();
    let second_id = second.id();
    slot.try_park(first).map_err(|_| ())?;

    let occupied = slot.try_park(second).err().ok_or(())?;
    assert_eq!(occupied.pending_id(), first_id);
    assert_eq!(occupied.into_rejected().id(), second_id);
    assert_eq!(slot.pending_id(), Some(first_id));
    Ok(())
}

#[test]
fn failed_default_admission_preserves_unissued_semantic_debt() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let mut slot = ContinuationSlot::new();
    let first = coordinator
        .try_admit(owed(1, 2), &mut slot)
        .map_err(|_| ())?;
    let first_id = first.into_parts().0.id();

    let failure = coordinator
        .try_admit(owed(3, 4), &mut slot)
        .err()
        .ok_or(())?;
    assert_eq!(failure.pending_id(), Some(first_id));
    let rejected = failure.into_owed();
    assert_eq!(rejected.work().0, 3);
    assert_eq!(rejected.timeout_after(), timeout_after());
    assert_eq!(slot.pending_id(), Some(first_id));
    Ok(())
}

#[test]
#[allow(clippy::clone_on_copy)]
fn public_value_and_recovery_surfaces_preserve_every_fact() -> Result<(), ()> {
    use core::hash::Hasher as _;

    let mut ids = ContinuationIds::<Machine>::default();
    let first_id = ids.try_issue().map_err(|_| ())?;
    let second_id = ids.try_issue().map_err(|_| ())?;
    let cloned_id = first_id.clone();
    assert_eq!(cloned_id, first_id);
    assert_eq!(format!("{first_id:?}"), "1");
    assert_eq!(first_id.partial_cmp(&second_id), Some(Ordering::Less));
    assert_eq!(first_id.cmp(&second_id), Ordering::Less);
    let mut first_hasher = DefaultHasher::new();
    first_id.hash(&mut first_hasher);
    let mut cloned_hasher = DefaultHasher::new();
    cloned_id.hash(&mut cloned_hasher);
    assert_eq!(first_hasher.finish(), cloned_hasher.finish());
    assert_eq!(timeout_after().nanoseconds().get(), 17);

    let mut coordinator = ContinuationCoordinator::new();
    let (work, deadline, continuation) = issue(&mut coordinator, 7, 11)?.into_parts();
    let work_id = work.id();
    let mapped = work.map(|Work(value)| value == 7);
    let (mapped_id, mapped_work) = mapped.into_parts();
    assert_eq!(mapped_id, work_id);
    assert!(mapped_work);
    let (deadline_id, duration) = deadline.into_parts();
    assert_eq!(deadline_id, work_id);
    assert_eq!(duration, timeout_after());
    let (_, cancellation) = continuation.complete(Completion(13)).into_parts();
    assert_eq!(cancellation.into_id(), work_id);
    Ok(())
}

#[test]
#[allow(clippy::clone_on_copy)]
fn error_value_traits_and_owned_recovery_cover_every_shape() -> Result<(), ()> {
    let mut coordinator = ContinuationCoordinator::new();
    let mut slot = ContinuationSlot::<Machine, Completion, Expiry>::default();
    let first = coordinator
        .try_admit(owed(1, 2), &mut slot)
        .map_err(|_| ())?;
    let first_id = first.into_parts().0.id();
    let failure = coordinator
        .try_admit(owed(3, 4), &mut slot)
        .err()
        .ok_or(())?;
    assert_eq!(failure.pending_id(), Some(first_id));
    let rejected = failure.into_owed();
    assert_eq!(rejected.work().0, 3);

    let (_, _, second) = issue(&mut coordinator, 3, 4)?.into_parts();
    let second_id = second.id();
    let different = slot.try_take(second_id).err().ok_or(())?;
    let cloned_different = different.clone();
    assert_eq!(different, cloned_different);
    assert!(format!("{different:?}").starts_with("Different"));
    assert_eq!(slot.try_take(first_id).map_err(|_| ())?.id(), first_id);

    let vacant = slot.try_take(first_id).err().ok_or(())?;
    let same_vacant = ContinuationNotPending::Vacant {
        requested: first_id,
    };
    assert_eq!(vacant, same_vacant);
    assert_ne!(vacant, different);
    assert_ne!(different, vacant);
    assert!(format!("{vacant:?}").starts_with("Vacant"));
    assert_eq!(slot.pending_id(), None);

    let completion = ContinuationCompletion::new(first_id, Completion(19));
    let failure = slot.try_complete(completion).err().ok_or(())?;
    let (completion, cause) = failure.into_parts();
    assert_eq!(completion.id(), first_id);
    assert_eq!(completion.completion().0, 19);
    assert_eq!(cause, vacant);
    Ok(())
}