liminal-server 0.25.4

use super::{
    AdmissionRequest, Arc, AuthorityState, CONTRACT_VERSION, ChannelMap, Counters, Decided,
    DecisionIndex, DecisionKey, GrantLog, Indexed, MappedChannel, MembershipAction,
    MembershipAuthority, MembershipConfig, MembershipCounts, MembershipProvider, MembershipRefusal,
    MembershipRequest, MembershipSubject, Mutex, MutexGuard, OnceLock, PageBounds, Resource,
    ServerError, admitted, bump, bump_by, channel_map, grant_stream, invalid, judged,
    validated_bounds,
};

/// The membership gate: the product mapping, the provider seam, the
/// authority seam and the bounded decision index (ACCESS-006 R1, R2, R6).
pub struct MembershipGate {
    pub(super) served: GrantLog,
    pub(super) workspace_kind: String,
    pub(super) channel_kind: String,
    pub(super) map: ChannelMap,
    pub(super) bounds: PageBounds,
    pub(super) max_decisions: usize,
    pub(super) connect_workspace: Option<String>,
    pub(super) provider: Arc<dyn MembershipProvider>,
    pub(super) authority: OnceLock<Arc<dyn MembershipAuthority>>,
    pub(super) stream: OnceLock<Arc<grant_stream::StreamState>>,
    pub(super) index: Mutex<DecisionIndex>,
    pub(super) counters: Counters,
}

impl std::fmt::Debug for MembershipGate {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("MembershipGate")
            .field("served", &self.served)
            .field("channels", &self.map.channels.len())
            .field("bounds", &self.bounds)
            .field("max_decisions", &self.max_decisions)
            .field("authority", &self.authority.get().is_some())
            .finish_non_exhaustive()
    }
}

impl MembershipGate {
    /// Builds the gate from the validated product mapping over `provider`.
    /// No authority is installed: until one is, every protected act refuses
    /// [`MembershipRefusal::AuthorityMissing`].
    ///
    /// # Errors
    /// [`ServerError::ConfigValidation`] naming the first invalid member: a
    /// contract other than [`CONTRACT_VERSION`], an empty or control-bearing
    /// kind, log identity or mapping member, equal workspace and channel
    /// kinds, a zero bound, no protected channel, a channel or resource
    /// mapped twice, a conversation naming an unlisted channel, or a
    /// `connect_workspace` the mapping does not list.
    pub fn new(
        config: &MembershipConfig,
        provider: Arc<dyn MembershipProvider>,
    ) -> Result<Self, ServerError> {
        let bounds = validated_bounds(config)?;
        let map = channel_map(config)?;
        if let Some(workspace) = &config.connect_workspace
            && !map.workspaces.contains(workspace)
        {
            return Err(invalid(
                "auth.lys.membership.connect_workspace is not a mapped workspace",
            ));
        }
        Ok(Self {
            served: GrantLog {
                identity: config.log_identity.clone(),
                epoch: config.log_epoch,
            },
            workspace_kind: config.workspace_kind.clone(),
            channel_kind: config.channel_kind.clone(),
            map,
            bounds,
            max_decisions: config.max_decisions,
            connect_workspace: config.connect_workspace.clone(),
            provider,
            authority: OnceLock::new(),
            stream: OnceLock::new(),
            index: Mutex::new(DecisionIndex::default()),
            counters: Counters::default(),
        })
    }

    /// Installs the ACCESS-005 grant authority once.
    ///
    /// # Errors
    /// Returns the authority back when one is already installed, so one
    /// owner cannot be swapped for another under live bindings.
    pub fn install_authority(
        &self,
        authority: Arc<dyn MembershipAuthority>,
    ) -> Result<(), Arc<dyn MembershipAuthority>> {
        self.authority.set(authority)
    }

    /// Installs the production stream authority before protected admission.
    ///
    /// # Errors
    /// Refuses an existing authority of another owner; it is never swapped live.
    pub fn install_stream_authority(&self) -> Result<Arc<grant_stream::StreamState>, ServerError> {
        if let Some(state) = self.stream.get() {
            return Ok(Arc::clone(state));
        }
        let state = grant_stream::StreamState::for_log(self.served.clone());
        self.install_authority(state.clone())
            .map_err(|_| invalid("membership_authority_already_installed"))?;
        self.stream
            .set(Arc::clone(&state))
            .map_err(|_| invalid("membership_stream_already_installed"))?;
        Ok(state)
    }

    /// The grant log this gate asks from.
    #[must_use]
    pub const fn served_log(&self) -> &GrantLog {
        &self.served
    }

    /// Whether `channel` is membership-protected.
    #[must_use]
    pub fn protects(&self, channel: &str) -> bool {
        self.map.channels.contains_key(channel)
    }

    /// The protected channel's placement, as the mapping states it.
    #[must_use]
    pub fn mapped(&self, channel: &str) -> Option<&MappedChannel> {
        self.map.channels.get(channel)
    }

    /// The protected channel a participant conversation is placed under, as
    /// the mapping states it; `None` is unmapped, never guessed.
    #[must_use]
    pub fn channel_for_conversation(&self, conversation: u64) -> Option<&str> {
        self.map
            .conversations
            .get(&conversation)
            .map(String::as_str)
    }

    /// Every count, as of now.
    #[must_use]
    pub fn counts(&self) -> MembershipCounts {
        self.counters.snapshot()
    }

    /// How many decisions the index retains.
    ///
    /// # Errors
    /// [`MembershipRefusal::IndexPoisoned`].
    pub fn retained(&self) -> Result<usize, MembershipRefusal> {
        Ok(self.lock()?.entries.len())
    }

    pub(super) fn lock(&self) -> Result<MutexGuard<'_, DecisionIndex>, MembershipRefusal> {
        bump(&self.counters.locks);
        self.index
            .lock()
            .map_err(|_| MembershipRefusal::IndexPoisoned)
    }

    /// The authority's revision and generation while it is `Ready` on the
    /// served log; its exact cause otherwise.
    pub(super) fn ready(&self) -> Result<(u64, u64), MembershipRefusal> {
        let authority = self
            .authority
            .get()
            .ok_or(MembershipRefusal::AuthorityMissing)?;
        match authority.state() {
            AuthorityState::Ready {
                log,
                revision,
                generation,
            } => {
                if log == self.served {
                    Ok((revision, generation))
                } else {
                    Err(MembershipRefusal::LogMismatch)
                }
            }
            AuthorityState::NotReady => Err(MembershipRefusal::NotReady),
            AuthorityState::Unavailable { cause } => Err(MembershipRefusal::Unavailable { cause }),
            AuthorityState::Gap { after, missing } => {
                Err(MembershipRefusal::Gap { after, missing })
            }
            AuthorityState::Reset => Err(MembershipRefusal::Reset),
        }
    }

    /// Rechecks the authority after the index or a cold answer was acquired.
    /// A reader can lose readiness while the provider's previous answer is in flight.
    pub(super) fn stream_fence(&self, generation: u64) -> Result<u64, MembershipRefusal> {
        let (revision, current) = self.ready()?;
        if current != generation {
            return Err(MembershipRefusal::ResumePending);
        }
        Ok(revision)
    }

    /// Decides `action` on `channel` for connection `pid`'s verified
    /// `holder`: [`Decided::NotProtected`] for a channel the mapping does
    /// not list; otherwise the current decision, from the index when it
    /// holds one at the authority's generation, asked of the provider once
    /// when it does not. The lock is never held across the ask.
    ///
    /// # Errors
    /// The exact [`MembershipRefusal`]: no verified holder, the authority's
    /// state, the index bound, the provider's refusal or failure, or an
    /// answer that is not the one asked.
    pub fn check(
        &self,
        pid: u64,
        holder: Option<&MembershipSubject>,
        channel: &str,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        let Some(mapped) = self.map.channels.get(channel) else {
            return Ok(Decided::NotProtected);
        };
        let decided = self.check_mapped(pid, holder, channel, mapped, action);
        if decided.is_err() {
            bump(&self.counters.refused);
        }
        decided
    }

    /// [`Self::check`] at a final data-release seam (ACCESS-006 R3), counted
    /// as a release check.
    ///
    /// # Errors
    /// As for [`Self::check`].
    pub fn check_release(
        &self,
        pid: u64,
        holder: Option<&MembershipSubject>,
        channel: &str,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        bump(&self.counters.release_checks);
        self.check(pid, holder, channel, action)
    }

    pub(super) fn check_mapped(
        &self,
        pid: u64,
        holder: Option<&MembershipSubject>,
        channel: &str,
        mapped: &MappedChannel,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        let holder = holder.ok_or(MembershipRefusal::HolderUnverified)?;
        let (revision, generation) = self.ready()?;
        let key = DecisionKey {
            subject: holder.clone(),
            channel: channel.to_owned(),
            action,
        };
        let (ticket, at_least) = {
            let mut index = self.lock()?;
            index.adopt(generation)?;
            if let Some(verdict) = index.known(&key, pid) {
                self.stream_fence(generation)?;
                drop(index);
                bump(&self.counters.local_lookups);
                return verdict.result();
            }
            let at_least = revision.max(index.applied);
            let ticket = index.reserve(&key, mapped, pid, self.max_decisions)?;
            drop(index);
            (ticket, at_least)
        };
        let asked = self.ask(holder, mapped, action, at_least);
        let mut index = self.lock()?;
        if let Err(refusal) = self.stream_fence(generation) {
            index.forget(&key, ticket);
            return Err(refusal);
        }
        match asked {
            Ok(verdict) => {
                let current = index.generation == Some(generation)
                    && index.settle(&key, ticket, verdict.clone());
                if !current {
                    // A change or a resume overtook this ask: its answer is not
                    // the current one, and nothing is released on it.
                    index.forget(&key, ticket);
                    return Err(MembershipRefusal::ChangeApplying);
                }
                drop(index);
                verdict.result()
            }
            Err(refusal) => {
                index.forget(&key, ticket);
                drop(index);
                Err(refusal)
            }
        }
    }

    pub(super) fn ask(
        &self,
        holder: &MembershipSubject,
        mapped: &MappedChannel,
        action: MembershipAction,
        at_least: u64,
    ) -> Result<Indexed, MembershipRefusal> {
        let request = MembershipRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: mapped.workspace.clone(),
            subject: holder.to_contract(),
            resource: mapped.resource.clone(),
            action: action.as_str().to_owned(),
            at_least,
        };
        bump(&self.counters.decisions_asked);
        let decision = self
            .provider
            .decide(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        judged(&request, &self.served, mapped, &decision)
    }

    /// Scoped guest admission at `Connect` (ACCESS-006 R1): when the mapping
    /// names a `connect_workspace`, the holder must hold a current grant
    /// somewhere in it. Admission opens no channel; every act after it is
    /// decided on its own. `Ok(None)` when no admission is configured.
    ///
    /// # Errors
    /// The authority's state, the provider's failure, a substituted or stale
    /// answer, or Lys's refusal by name (`membership_no_current_grant`).
    pub fn admit_connect(
        &self,
        holder: &MembershipSubject,
    ) -> Result<Option<u64>, MembershipRefusal> {
        let Some(workspace) = &self.connect_workspace else {
            return Ok(None);
        };
        let (revision, generation) = self.ready()?;
        let request = AdmissionRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: Resource {
                kind: self.workspace_kind.clone(),
                id: workspace.clone(),
            },
            subject: holder.to_contract(),
            at_least: revision,
        };
        bump(&self.counters.admissions_asked);
        let decision = self
            .provider
            .admission(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        let admitted_at = admitted(&request, &self.served, &decision)?;
        let current = self.stream_fence(generation)?;
        if admitted_at < current {
            return Err(MembershipRefusal::RevisionStale {
                required: current,
                decided: Some(admitted_at),
            });
        }
        Ok(Some(admitted_at))
    }

    /// Releases every binding connection `pid` held (ACCESS-006 R6); a
    /// binding no connection holds leaves the index. A poisoned index is
    /// logged and left: it already refuses every check.
    pub fn release_connection(&self, pid: u64) {
        match self.lock() {
            Ok(mut index) => {
                let released = index.release(pid);
                drop(index);
                bump_by(&self.counters.released, released);
            }
            Err(refusal) => {
                tracing::warn!(
                    pid,
                    refusal = refusal.name(),
                    "membership bindings could not be released"
                );
            }
        }
    }

    /// Re-asks every key in `tickets` at `at_least`, settling each; returns
    /// how many settled and how many failed and were released.
    pub(super) fn redecide(&self, tickets: Vec<(DecisionKey, u64)>, at_least: u64) -> (u64, u64) {
        let mut settled = 0_u64;
        let mut failed = 0_u64;
        for (key, ticket) in tickets {
            let asked = self
                .map
                .channels
                .get(&key.channel)
                .ok_or(MembershipRefusal::ChannelNotMapped)
                .and_then(|mapped| self.ask(&key.subject, mapped, key.action, at_least));
            let Ok(mut index) = self.lock() else {
                failed = failed.saturating_add(1);
                continue;
            };
            let kept = asked.is_ok_and(|verdict| index.settle(&key, ticket, verdict));
            if kept {
                settled = settled.saturating_add(1);
            } else {
                index.forget(&key, ticket);
                failed = failed.saturating_add(1);
            }
        }
        (settled, failed)
    }
}