pipecircuit 0.0.15

use crate::storage::warp_table::{
    Capacity, WarpId, WarpTable, WarpTableCreationError, WarpTableInsertOutcome,
    WarpTableRemoveOutcome, WarpTableRow,
};
use crate::{Participant, TimerToken};

use super::{
    CancelledContinuationCompletion, ContinuationAdmission, ContinuationArmRefusal,
    ContinuationCompletion, ContinuationFailure, ContinuationId, ContinuationResult,
    ContinuationUnavailable, Owed,
};

#[derive(WarpTableRow)]
struct PendingContinuation<Address> {
    return_to: Address,
}

type PendingContinuationCreationError<Address> = WarpTableCreationError<
    <PendingContinuation<Address> as WarpTableRow>::ColumnsCreationError,
    <PendingContinuation<Address> as WarpTableRow>::IndexesCreationError,
>;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ContinuationLedgerCreationFailure {
    StorageUnavailable,
}

#[derive(Debug)]
pub struct ContinuationLedgerCreationError<Address> {
    failure: ContinuationLedgerCreationFailure,
    _source: PendingContinuationCreationError<Address>,
}

impl<Address> ContinuationLedgerCreationError<Address> {
    pub const fn failure(&self) -> ContinuationLedgerCreationFailure {
        self.failure
    }
}

pub struct ContinuationLedger<Address> {
    pending: WarpTable<PendingContinuation<Address>>,
}

impl<Address> ContinuationLedger<Address> {
    pub fn try_new(capacity: Capacity) -> Result<Self, ContinuationLedgerCreationError<Address>> {
        let pending =
            WarpTable::try_new(capacity).map_err(|source| ContinuationLedgerCreationError {
                failure: ContinuationLedgerCreationFailure::StorageUnavailable,
                _source: source,
            })?;
        Ok(Self { pending })
    }

    pub const fn capacity(&self) -> Capacity {
        self.pending.capacity()
    }

    pub const fn len(&self) -> usize {
        self.pending.len()
    }

    pub const fn is_empty(&self) -> bool {
        self.pending.is_empty()
    }

    pub fn admit<'message, Worker>(
        &mut self,
        return_to: Address,
        owed: Owed<Worker::In<'message>>,
    ) -> ContinuationAdmissionOutcome<'message, Address, Worker>
    where
        Worker: Participant,
    {
        match self.pending.insert(PendingContinuation { return_to }) {
            WarpTableInsertOutcome::Inserted { id, .. } => ContinuationAdmissionOutcome::Admitted {
                admission: ContinuationAdmission {
                    id: ContinuationId::new(id.value()),
                    owed,
                },
            },
            WarpTableInsertOutcome::Full {
                rejected: PendingContinuation { return_to },
            } => ContinuationAdmissionOutcome::Refused {
                return_to,
                result: Err(ContinuationFailure::Unavailable {
                    reason: ContinuationUnavailable::AtCapacity {
                        maximum_pending: self.capacity().value().get(),
                    },
                    rejected: owed,
                }),
            },
            WarpTableInsertOutcome::IdentifiersExhausted {
                rejected: PendingContinuation { return_to },
            } => ContinuationAdmissionOutcome::Refused {
                return_to,
                result: Err(ContinuationFailure::Unavailable {
                    reason: ContinuationUnavailable::IdentifiersExhausted,
                    rejected: owed,
                }),
            },
            WarpTableInsertOutcome::UniqueConflict {
                conflict,
                rejected: _,
            } => match conflict {},
        }
    }

    pub fn resolve<'message, Worker>(
        &mut self,
        completion: CancelledContinuationCompletion<'message, Worker>,
    ) -> ContinuationResolutionOutcome<'message, Address, Worker>
    where
        Worker: Participant,
    {
        let completion = completion.0;
        let id = WarpId::new(completion.id.value());
        match self.pending.remove(id) {
            WarpTableRemoveOutcome::Removed {
                removed: PendingContinuation { return_to },
                ..
            } => ContinuationResolutionOutcome::Resolved {
                return_to,
                result: Ok(completion.output),
            },
            WarpTableRemoveOutcome::Absent { .. } => {
                ContinuationResolutionOutcome::Stale { completion }
            }
        }
    }

    pub fn refuse<'message, Worker>(
        &mut self,
        refusal: ContinuationArmRefusal<'message, Worker>,
    ) -> ContinuationRefusalOutcome<'message, Address, Worker>
    where
        Worker: Participant,
    {
        let id = WarpId::new(refusal.id.value());
        match self.pending.remove(id) {
            WarpTableRemoveOutcome::Removed {
                removed: PendingContinuation { return_to },
                ..
            } => ContinuationRefusalOutcome::Refused {
                return_to,
                result: Err(ContinuationFailure::Unavailable {
                    reason: ContinuationUnavailable::TimerRefused {
                        reason: refusal.reason,
                    },
                    rejected: refusal.owed,
                }),
            },
            WarpTableRemoveOutcome::Absent { .. } => ContinuationRefusalOutcome::Stale { refusal },
        }
    }

    pub fn expire(&mut self, token: TimerToken) -> ContinuationExpirationOutcome<Address> {
        let id = WarpId::new(token.value());
        match self.pending.remove(id) {
            WarpTableRemoveOutcome::Removed {
                removed: PendingContinuation { return_to },
                ..
            } => ContinuationExpirationOutcome::Expired {
                return_to,
                expiration: ContinuationExpiration { token },
            },
            WarpTableRemoveOutcome::Absent { .. } => ContinuationExpirationOutcome::Stale { token },
        }
    }
}

#[must_use = "admitted work or immediate refusal must be handled by the fitting"]
pub enum ContinuationAdmissionOutcome<'message, Address, Worker>
where
    Worker: Participant,
{
    Admitted {
        admission: ContinuationAdmission<'message, Worker>,
    },
    Refused {
        return_to: Address,
        result: ContinuationResult<'message, Worker>,
    },
}

#[must_use = "resolved or stale completion must be handled by the fitting"]
pub enum ContinuationResolutionOutcome<'message, Address, Worker>
where
    Worker: Participant,
{
    Resolved {
        return_to: Address,
        result: ContinuationResult<'message, Worker>,
    },
    Stale {
        completion: ContinuationCompletion<'message, Worker>,
    },
}

#[must_use = "resolved or stale timer refusal must be handled by the fitting"]
pub enum ContinuationRefusalOutcome<'message, Address, Worker>
where
    Worker: Participant,
{
    Refused {
        return_to: Address,
        result: ContinuationResult<'message, Worker>,
    },
    Stale {
        refusal: ContinuationArmRefusal<'message, Worker>,
    },
}

#[must_use = "expired or stale timer tokens must be handled by the fitting"]
pub enum ContinuationExpirationOutcome<Address> {
    Expired {
        return_to: Address,
        expiration: ContinuationExpiration,
    },
    Stale {
        token: TimerToken,
    },
}

#[must_use = "expiration must become the continued pipe result"]
pub struct ContinuationExpiration {
    token: TimerToken,
}

impl ContinuationExpiration {
    pub const fn token(&self) -> TimerToken {
        self.token
    }

    pub const fn into_result<'message, Worker>(self) -> ContinuationResult<'message, Worker>
    where
        Worker: Participant,
    {
        Err(ContinuationFailure::Expired)
    }
}