liminal-server 0.25.4

use super::{AtomicU64, GrantLog, MembershipRefusal, Ordering, Resource, Subject};

/// The ACCESS-005 grant authority's state, as the membership gate reads it
/// (ACCESS-006 R2).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AuthorityState {
    /// No verified baseline yet.
    NotReady,
    /// Baseline, cursor and indexes agree on `log` at `revision`, in resume
    /// generation `generation`.
    Ready {
        /// The grant log the authority is current on.
        log: GrantLog,
        /// The last applied receipt revision.
        revision: u64,
        /// The settled resume generation.
        generation: u64,
    },
    /// The change feed or the provider is lost.
    Unavailable {
        /// The stable cause.
        cause: String,
    },
    /// The change feed has an uncovered interval.
    Gap {
        /// The last revision applied before the gap.
        after: u64,
        /// How many revisions are missing.
        missing: u64,
    },
    /// The grant log was reset and the new baseline is not yet verified.
    Reset,
}

/// The seam ACCESS-005's versioned grant owner implements for membership
/// (ACCESS-006 R2): its current state, read locally with no network call and
/// no wait.
///
/// The one owner per grant log drives ordered changes and resumes
/// into the gate through [`MembershipListener`]; there is no second feed.
pub trait MembershipAuthority: Send + Sync + std::fmt::Debug {
    /// The authority's current state.
    fn state(&self) -> AuthorityState;
}

/// One applied grant change, in the owner's order.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct MembershipChange {
    /// The grant log the change belongs to.
    pub log: GrantLog,
    /// Its receipt revision.
    pub revision: u64,
    /// The subject whose grant changed; `None` when every holder on the
    /// resource is affected.
    pub subject: Option<Subject>,
    /// The resource the changed grant is held on: a channel or a workspace.
    pub resource: Resource,
}

/// What applying one change did.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct AppliedChange {
    /// The revision was already applied: nothing was touched.
    pub duplicate: bool,
    /// Retained bindings the change's resource reaches.
    pub affected: u64,
    /// Of those, re-decided at the change's revision.
    pub redecided: u64,
    /// Of those, released because the provider could not answer; the next
    /// act asks again.
    pub failed: u64,
}

/// What one resume did.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ResumeReport {
    /// Retained bindings re-decided once for the generation.
    pub rechecked: u64,
    /// Of those, released because the provider could not answer.
    pub failed: u64,
    /// Whether every binding agreed and the gate adopted the generation:
    /// the owner republishes `Ready` only when this holds.
    pub settled: bool,
}

/// The gate's side of the ACCESS-005 owner's ordered consumer.
///
/// The owner
/// calls [`Self::changed`] after applying each change and before it admits
/// any later act, and [`Self::resumed`] after a resume's contiguous tail and
/// before it republishes `Ready`.
pub trait MembershipListener: Send + Sync {
    /// Re-decides the retained bindings `change` reaches.
    ///
    /// # Errors
    /// [`MembershipRefusal::LogMismatch`] for another log; the index's own
    /// refusals.
    fn changed(&self, change: &MembershipChange) -> Result<AppliedChange, MembershipRefusal>;

    /// Re-decides every retained binding once at `revision` and adopts
    /// `generation` when all agree.
    ///
    /// # Errors
    /// The index's own refusals.
    fn resumed(&self, generation: u64, revision: u64) -> Result<ResumeReport, MembershipRefusal>;
}

/// The result of a membership check.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Decided {
    /// The channel is not membership-protected: its prior contract applies.
    NotProtected,
    /// Allowed by `grant`, decided at `revision`.
    Allowed {
        /// The granting receipt.
        grant: String,
        /// The revision decided at.
        revision: u64,
    },
}

/// Every count the gate keeps, by owned phase (ACCESS-006 R6). There is no
/// history, roster or whole-state structure to traverse or clone, so no such
/// count exists; a page's rows are counted as received.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct MembershipCounts {
    /// Membership decisions asked of the provider.
    pub decisions_asked: u64,
    /// Admission decisions asked of the provider.
    pub admissions_asked: u64,
    /// Pages asked of the provider.
    pub pages_asked: u64,
    /// Page rows received.
    pub page_rows: u64,
    /// Checks answered from the index.
    pub local_lookups: u64,
    /// Release checks made at a delivery seam.
    pub release_checks: u64,
    /// Refused checks.
    pub refused: u64,
    /// Bindings a change's reverse index visited.
    pub affected_visited: u64,
    /// Bindings a resume re-decided.
    pub resume_rechecks: u64,
    /// Bindings released when their last connection closed.
    pub released: u64,
    /// Index lock acquisitions.
    pub locks: u64,
}

#[derive(Debug, Default)]
pub(super) struct Counters {
    pub(super) decisions_asked: AtomicU64,
    pub(super) admissions_asked: AtomicU64,
    pub(super) pages_asked: AtomicU64,
    pub(super) page_rows: AtomicU64,
    pub(super) local_lookups: AtomicU64,
    pub(super) release_checks: AtomicU64,
    pub(super) refused: AtomicU64,
    pub(super) affected_visited: AtomicU64,
    pub(super) resume_rechecks: AtomicU64,
    pub(super) released: AtomicU64,
    pub(super) locks: AtomicU64,
}

impl Counters {
    pub(super) fn snapshot(&self) -> MembershipCounts {
        let read = |counter: &AtomicU64| counter.load(Ordering::Relaxed);
        MembershipCounts {
            decisions_asked: read(&self.decisions_asked),
            admissions_asked: read(&self.admissions_asked),
            pages_asked: read(&self.pages_asked),
            page_rows: read(&self.page_rows),
            local_lookups: read(&self.local_lookups),
            release_checks: read(&self.release_checks),
            refused: read(&self.refused),
            affected_visited: read(&self.affected_visited),
            resume_rechecks: read(&self.resume_rechecks),
            released: read(&self.released),
            locks: read(&self.locks),
        }
    }
}