pipecircuit 0.0.18

use core::num::{NonZeroU32, NonZeroU64};

use crate::{Pipe, TimerArmRefusal, TimerIn, TimerOut, TimerToken, storage::warp_table::Capacity};

use super::*;

const fn wake_after() -> WakeAfter {
    WakeAfter::milliseconds(NonZeroU64::MIN)
}

fn ledger(capacity: u32) -> Result<DelayLedger<u8>, &'static str> {
    let capacity = Capacity::new(NonZeroU32::new(capacity).ok_or("capacity was zero")?);
    DelayLedger::try_new(capacity).map_err(|_| "delay storage unavailable")
}

fn admit(ledger: &mut DelayLedger<u8>, message: u8) -> Result<DelayAdmission, &'static str> {
    match ledger.admit(Delayed::new(message, wake_after())) {
        DelayAdmissionOutcome::Admitted { admission } => Ok(admission),
        DelayAdmissionOutcome::Refused { .. } => Err("delayed message was refused"),
    }
}

fn arm(admission: DelayAdmission) -> Result<(), &'static str> {
    let TimerIn::Arm { token, .. } = admission.timer_input() else {
        return Err("delay admission did not arm a timer");
    };
    let outcome = TimerOut::Armed { token }
        .into_arm_outcome()
        .map_err(|_| "timer arm output was rejected")?;
    match admission
        .settle_timer_arm(outcome)
        .map_err(|_| "timer returned the wrong token")?
    {
        DelayArmOutcome::Armed => Ok(()),
        DelayArmOutcome::Refused { .. } => Err("timer refused delayed work"),
    }
}

#[test]
fn after_participant_accepts_delayed_and_emits_the_owned_message() {
    fn accepts_input(_input: <After<u8> as Pipe>::In<'_>) {}
    fn accepts_output(_output: <After<u8> as Pipe>::Out<'_>) {}

    let delayed = Delayed::new(7, wake_after());
    assert_eq!(delayed.message(), &7);
    assert_eq!(delayed.wake_after(), wake_after());
    accepts_input(delayed);
    accepts_output(7);
}

#[test]
fn armed_message_remains_pending_until_one_elapsed_token_releases_it() -> Result<(), &'static str> {
    let mut ledger = ledger(1)?;
    let admission = admit(&mut ledger, 7)?;
    let TimerIn::Arm { token, after } = admission.timer_input() else {
        return Err("delay admission did not arm a timer");
    };
    assert_eq!(after, wake_after());
    arm(admission)?;
    assert_eq!(ledger.len(), 1);
    assert!(!ledger.is_empty());
    assert!(matches!(
        ledger.release(token),
        DelayReleaseOutcome::Released { message: 7 }
    ));
    assert!(matches!(
        ledger.release(token),
        DelayReleaseOutcome::Stale { token: stale } if stale == token
    ));
    assert!(ledger.is_empty());
    Ok(())
}

#[test]
fn capacity_refusal_preserves_the_message_and_wake_precision() -> Result<(), &'static str> {
    let mut ledger = ledger(1)?;
    let _first = admit(&mut ledger, 1)?;
    let DelayAdmissionOutcome::Refused { reason, rejected } =
        ledger.admit(Delayed::new(2, wake_after()))
    else {
        return Err("full delay ledger admitted another message");
    };
    assert_eq!(reason, DelayUnavailable::AtCapacity { maximum_pending: 1 });
    assert_eq!(rejected.message(), &2);
    assert_eq!(rejected.wake_after(), wake_after());
    Ok(())
}

#[test]
fn timer_refusal_returns_the_complete_delayed_message() -> Result<(), &'static str> {
    let mut ledger = ledger(1)?;
    let admission = admit(&mut ledger, 9)?;
    let TimerIn::Arm { token, after } = admission.timer_input() else {
        return Err("delay admission did not arm a timer");
    };
    let reason = TimerArmRefusal::AtCapacity {
        maximum_deadlines: 1,
    };
    let outcome = TimerOut::ArmRefused {
        token,
        after,
        reason,
    }
    .into_arm_outcome()
    .map_err(|_| "timer refusal output was rejected")?;
    let DelayArmOutcome::Refused { refusal } = admission
        .settle_timer_arm(outcome)
        .map_err(|_| "timer returned the wrong token")?
    else {
        return Err("refused delayed message was armed");
    };
    assert_eq!(refusal.reason(), reason);
    let DelayRefusalOutcome::Refused {
        reason: unavailable,
        rejected,
    } = ledger.refuse(refusal)
    else {
        return Err("timer refusal was stale");
    };
    assert_eq!(unavailable, DelayUnavailable::WakeRefused { reason });
    assert_eq!(rejected.message(), &9);
    assert_eq!(rejected.wake_after(), wake_after());
    assert!(ledger.is_empty());
    Ok(())
}

#[test]
fn mismatched_timer_token_preserves_both_sides() -> Result<(), &'static str> {
    let mut ledger = ledger(1)?;
    let admission = admit(&mut ledger, 5)?;
    let TimerIn::Arm { token, .. } = admission.timer_input() else {
        return Err("delay admission did not arm a timer");
    };
    let wrong = TimerToken::from_nonzero(
        token
            .value()
            .checked_add(1)
            .ok_or("timer token overflowed")?,
    );
    let outcome = TimerOut::Armed { token: wrong }
        .into_arm_outcome()
        .map_err(|_| "timer arm output was rejected")?;
    let mismatch = admission
        .settle_timer_arm(outcome)
        .err()
        .ok_or("wrong timer token was accepted")?;
    let (admission, decision) = mismatch.into_parts();
    assert!(matches!(decision, crate::TimerArmDecision::Armed { token } if token == wrong));
    let TimerIn::Arm {
        token: preserved, ..
    } = admission.timer_input()
    else {
        return Err("preserved admission did not arm a timer");
    };
    assert_eq!(preserved, token);
    let DelayWithdrawalOutcome::Withdrawn { rejected } = ledger.withdraw(admission) else {
        return Err("mismatched admission disappeared");
    };
    assert_eq!(rejected.message(), &5);
    assert_eq!(rejected.wake_after(), wake_after());
    assert!(ledger.is_empty());
    Ok(())
}