liminal-server 0.25.4

//! Atomic target receipts accompany explicit, authenticated connector pulls.
//!
//! Connector pages omit approving people; their authority is recorded as `lys`
//! with the draft id, never as a person inferred from the requesting holder.

use std::collections::BTreeSet;
use std::num::NonZeroUsize;

use lys_pass::drafts::{ApprovedDraft, Execution, Executor};
use serde::{Deserialize, Serialize};

#[path = "../http/admin_drafts.rs"]
pub mod admin;
#[path = "grant_stream.rs"]
mod pulls;
#[path = "../agents/waiting.rs"]
pub mod waiting;
pub use admin::DraftPuller;
pub use projection::AgentProjection;
pub use pulls::AgentDraftStore;
pub use target::TargetDraftStore;
#[path = "drafts_projection.rs"]
mod projection;
#[path = "drafts_target.rs"]
mod target;

/// Explicit resource limits for a single requested pull.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DraftLimits {
    /// Bounds pages traversed by one caller so a moving population cannot
    /// hold the administration lane indefinitely. No timer ends a pull.
    pub pages: NonZeroUsize,
    /// Bounds distinct draft ids handled per pull, including replays; nothing is evicted
    /// from the durable receipt ledger when the limit is reached.
    pub drafts: NonZeroUsize,
    /// Bounds authenticated approver identities before the transition is kept;
    /// this must accommodate the grant's required number of distinct people.
    pub approvers: NonZeroUsize,
    /// Bounds exact immutable words before any execution or retained-state write.
    pub words_bytes: NonZeroUsize,
    /// Bounds ids, grant ids, connector credentials and opaque page cursors.
    pub metadata_bytes: NonZeroUsize,
}

/// Approval evidence already authenticated by Lys, bound to the exact words.
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ApprovalEvidence {
    /// The immutable draft whose approvals were read from Lys.
    pub draft: String,
    /// The immutable request approved by the named people.
    pub digest: String,
    /// Actual approvers when served, otherwise the explicit issuer marker `lys`.
    pub approved_by: Vec<String>,
    /// Prepared held mode; an issuer-only refusal uses one authority marker.
    pub mode: super::super::GrantMode,
}

/// Stored in the same durable operation as the target mutation and receipt.
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ApprovalTransition {
    /// The Lys draft that authorized this transition.
    pub draft: String,
    /// SHA-256 of the exact approved words.
    pub digest: String,
    /// Served approving people or `lys`; the accompanying draft identifies the issuer evidence.
    pub approved_by: Vec<String>,
    /// Milliseconds since the epoch, supplied by the server's operation clock.
    pub executed_at_ms: u64,
}

/// The existing registry's atomic target interface; it owns all prepared bytes.
///
/// Implementations must bind draft id, grant, kind, id, action and digest to the
/// stored prepared act. They revalidate the current target and exact words under
/// the commit hold, and store a named refusal for missing/retired targets.
/// Each owning registry executes its typed prepared command through this seam.
/// The human words never select an unapproved dispatcher or command.
pub trait AtomicDraftStore: Send {
    /// Reads a committed receipt by immutable draft id.
    ///
    /// # Errors
    /// Durable ledger read failure.
    fn receipt(&self, draft: &str) -> Result<Option<Execution>, DraftError>;

    /// Reads authenticated approver evidence for this exact draft and digest.
    ///
    /// # Errors
    /// Missing evidence or durable read failure; the act remains unapplied.
    fn approval(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError>;

    /// One durable operation applies the prepared target mutation, transition,
    /// receipt and outcome outbox, or commits a named refusal without mutation.
    /// Concurrent/replayed ids answer the first receipt. A stored digest,
    /// target, grant or words mismatch never applies a new mutation.
    ///
    /// # Errors
    /// Storage or execution failure. Business refusals are durable `Execution`
    /// values so another pull never attempts the rejected act again.
    fn commit(
        &mut self,
        draft: &ApprovedDraft,
        transition: &ApprovalTransition,
    ) -> Result<Execution, DraftError>;

    /// Whether Lys acknowledged this receipt; the marker survives restart.
    ///
    /// # Errors
    /// Durable ledger read failure.
    fn acknowledged(&self, draft: &str) -> Result<bool, DraftError>;

    /// Keeps the successful acknowledgment without changing the target again.
    /// A marker can be set only when `receipt` matches the committed outcome.
    ///
    /// # Errors
    /// Storage failure or a mismatching receipt.
    fn acknowledge(&mut self, draft: &str, receipt: &Execution) -> Result<(), DraftError>;

    /// Reads at most `limit` locally committed but unacknowledged drafts.
    /// A close whose response was lost may already be absent from Lys's page.
    ///
    /// # Errors
    /// Durable outbox read failure; no uncommitted act is returned here.
    fn pending_acknowledgments(
        &self,
        limit: NonZeroUsize,
    ) -> Result<Vec<ApprovedDraft>, DraftError>;

    /// Delivers the committed outbox through the retained state writer and
    /// inbox with server provenance. The same operation is idempotent after
    /// restart. An unrelated holder is never rebound from a request's name.
    ///
    /// # Errors
    /// State or inbox delivery failure; the receipt remains committed.
    fn deliver_outcome(
        &mut self,
        draft: &ApprovedDraft,
        receipt: &Execution,
    ) -> Result<(), DraftError>;
}

/// Named failures contain no token, prepared words or HTTP response body.
#[derive(Debug, thiserror::Error)]
pub enum DraftError {
    /// Lys's shared consumer refused the HTTP contract by its own name.
    #[error("draft_lys_failed: {0}")]
    Lys(String),
    /// A configured positive byte, page or count bound was exceeded.
    #[error("draft_bound_exceeded")]
    Bound,
    /// A continuation revisited an earlier cursor.
    #[error("draft_cursor_repeated")]
    CursorRepeated,
    /// The same id was presented again in one pull.
    #[error("draft_id_repeated")]
    IdRepeated,
    /// No authenticated approver evidence can be read for these exact words.
    #[error("draft_approval_evidence_missing")]
    ApprovalUnavailable,
    /// The durable receipt describes another immutable request.
    #[error("draft_receipt_digest_mismatch")]
    ReceiptMismatch,
    /// A target or receipt could not be read, committed or published.
    #[error("draft_store_failed")]
    Store,
    /// Startup has already signalled its one pull.
    #[error("draft_startup_pull_repeated")]
    StartupRepeated,
    /// Lys could not acknowledge a receipt that is already committed.
    #[error("draft_acknowledgment_failed: {draft}: {name}")]
    Acknowledgment {
        /// The draft whose local receipt remains committed.
        draft: String,
        /// The issuer consumer's safe machine failure name.
        name: String,
    },
    /// The shared consumer refused the immutable execution contract.
    #[error("draft_contract_refused")]
    Contract,
}

/// The only events which request an approved-draft pull.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PullSignal {
    /// Once, after the target registries are loaded.
    Startup,
    /// A person's console approval has returned from Lys.
    Console,
    /// The stream owner delivered an explicit draft decision row.
    Decision,
}

/// A future stream row is recognized by its declared name; unrelated grant
/// changes never imply a draft approval. Lys currently publishes no such row.
#[must_use]
pub fn stream_signal(kind: &str) -> Option<PullSignal> {
    (kind == "draft_decision").then_some(PullSignal::Decision)
}

/// Join for the one grant-stream owner; no thread, timer or second reader.
#[async_trait::async_trait]
pub trait DraftDecisionHook: Send + Sync {
    /// Requests one pull after the owner accepts a named draft decision row.
    ///
    /// # Errors
    /// The explicit pull's named failure; the owner retains it in diagnostics.
    async fn draft_decided(&self) -> Result<PullReport, DraftError>;
}

/// Counts from one explicit pull, without carrying draft words or credentials.
#[derive(Default, Debug, Serialize)]
pub struct PullReport {
    /// Pages answered by Lys.
    pub pages: usize,
    /// New executed receipts.
    pub executed: usize,
    /// New refused receipts.
    pub refused: usize,
    /// Previously committed receipts returned without another mutation.
    pub replayed: usize,
    /// Outcomes newly acknowledged by Lys.
    pub acknowledged: usize,
}

/// Lys's executor over the existing target owner's atomic operation.
pub struct DraftExecutor<S> {
    pub(crate) store: S,
    pub(crate) limits: DraftLimits,
    pub(crate) at_ms: u64,
}

impl<S: AtomicDraftStore> Executor for DraftExecutor<S> {
    type Error = DraftError;

    fn receipt(&self, draft: &str) -> Result<Option<Execution>, DraftError> {
        self.store.receipt(draft)
    }

    fn execute_and_record(&mut self, draft: &ApprovedDraft) -> Result<Execution, DraftError> {
        let evidence = self.store.approval(draft)?;
        let mut people = BTreeSet::new();
        let needed = match evidence.mode {
            super::super::GrantMode::ByDraft => 1,
            super::super::GrantMode::ByTwo => 2,
            super::super::GrantMode::Outright => return Err(DraftError::ApprovalUnavailable),
        };
        if evidence.draft != draft.id
            || evidence.digest != draft.request_digest
            || evidence.approved_by.len() != needed
            || evidence.approved_by.len() > self.limits.approvers.get()
            || evidence.approved_by.iter().any(|person| {
                !valid_metadata(person, self.limits.metadata_bytes) || !people.insert(person)
            })
        {
            return Err(DraftError::ApprovalUnavailable);
        }
        self.store.commit(
            draft,
            &ApprovalTransition {
                draft: draft.id.clone(),
                digest: draft.request_digest.clone(),
                approved_by: evidence.approved_by,
                executed_at_ms: self.at_ms,
            },
        )
    }
}

pub(crate) fn valid_metadata(value: &str, bound: NonZeroUsize) -> bool {
    !value.is_empty() && value.len() <= bound.get() && !value.chars().any(char::is_control)
}

pub(crate) fn validate_bounds(
    draft: &ApprovedDraft,
    limits: DraftLimits,
) -> Result<(), DraftError> {
    bounded(
        &draft.id,
        &draft.app,
        &draft.grant,
        &draft.target,
        &draft.words,
        limits,
    )
}

/// The same bounds for a request Lys holds, before any approval exists.
pub(crate) fn validate_request_bounds(
    request: &lys_pass::drafts::DraftRequest,
    limits: DraftLimits,
) -> Result<(), DraftError> {
    bounded(
        &request.operation,
        "liminal",
        &request.grant,
        &request.target,
        &request.words,
        limits,
    )
}

fn bounded(
    id: &str,
    app: &str,
    grant: &str,
    target: &lys_pass::Target,
    words: &str,
    limits: DraftLimits,
) -> Result<(), DraftError> {
    if words.is_empty()
        || words.len() > limits.words_bytes.get()
        || [id, app, grant, &target.kind, &target.id, &target.action]
            .iter()
            .any(|value| !valid_metadata(value, limits.metadata_bytes))
    {
        return Err(DraftError::Bound);
    }
    Ok(())
}

/// The immutable digest shared by either terminal outcome.
#[must_use]
pub fn execution_digest(receipt: &Execution) -> &str {
    match receipt {
        Execution::Executed { request_digest, .. }
        | Execution::RefusedOnExecution { request_digest, .. } => request_digest,
    }
}

/// Compares every immutable member before a prepared target is applied.
#[must_use]
pub fn same_request(request: &lys_pass::drafts::DraftRequest, draft: &ApprovedDraft) -> bool {
    draft.app == "liminal"
        && request.grant == draft.grant
        && request.target == draft.target
        && request.request_digest == draft.request_digest
        && request.words == draft.words
}

/// Approval evidence from the authenticated connector page. The issuer marker
/// records the limitation of that page; it does not name a person as approver.
///
/// # Errors
/// An unavailable prepared ledger or invalid held mode.
pub fn issuer_approval(
    journal: &waiting::durable::DraftJournal,
    draft: &ApprovedDraft,
) -> Result<ApprovalEvidence, DraftError> {
    // Lys names every approver and the receipt's mode on the approved draft
    // itself (lys-pass 0.3.4), bound to its id and digest. A preparation this
    // exchange recorded must agree with that mode; an unknown preparation is
    // judged by the issuer's evidence alone.
    let mode = match draft.mode {
        lys_pass::Mode::Outright => super::super::GrantMode::Outright,
        lys_pass::Mode::ByDraft => super::super::GrantMode::ByDraft,
        lys_pass::Mode::ByTwo => super::super::GrantMode::ByTwo,
    };
    let prepared = journal.read(|snapshot| {
        snapshot
            .prepared
            .values()
            .find(|prepared| prepared.draft.as_deref() == Some(&draft.id))
            .map(|prepared| prepared.mode)
    })?;
    if prepared.is_some_and(|prepared| prepared != mode) {
        return Err(DraftError::ApprovalUnavailable);
    }
    Ok(ApprovalEvidence {
        draft: draft.id.clone(),
        digest: draft.request_digest.clone(),
        approved_by: draft.approved_by.clone(),
        mode,
    })
}

impl crate::services::lifecycle::site::executor::ApprovalReader for waiting::durable::DraftJournal {
    fn read(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError> {
        issuer_approval(self, draft)
    }
}

/// Completes one synchronous permission call without nesting a runtime.
pub(super) fn block<T, F>(handle: &tokio::runtime::Handle, future: F) -> Result<T, lys_pass::Error>
where
    T: Send,
    F: std::future::Future<Output = Result<T, lys_pass::Error>> + Send,
{
    std::thread::scope(|scope| scope.spawn(move || handle.block_on(future)).join())
        .unwrap_or_else(|_| Err(lys_pass::Error::CannotAsk("the live ask did not complete")))
}

pub(super) fn prepared_failure(_: DraftError) -> lys_pass::Error {
    lys_pass::Error::CannotAsk("draft target journal failed")
}

/// Explicit capacity for durable evidence, which is never evicted to admit work.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct JournalLimits {
    /// Bounds native target history and migration allocation before admission.
    pub rows: NonZeroUsize,
    /// Bounds prepared operations and receipts retained for immutable replays.
    pub operations: NonZeroUsize,
    /// Bounds the whole atomic request before it reaches the durable engine.
    pub snapshot_bytes: NonZeroUsize,
    /// Bounds each read of an old installation during its one-time migration.
    pub replay_page_rows: NonZeroUsize,
}

impl AgentDraftStore {
    /// Shares the server's existing native inbox and retirement projection.
    #[must_use]
    pub fn with_projection(mut self, projection: std::sync::Arc<dyn AgentProjection>) -> Self {
        self.projection = projection;
        self
    }
}

/// Operator-selected admission, journal, retained-history and HTTP capacities.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DraftConfig {
    pub pull: DraftLimits,
    pub journal: JournalLimits,
    pub waiting: waiting::WaitingLimits,
    pub http: admin::AdminLimits,
}

#[cfg(test)]
#[path = "drafts_tests.rs"]
pub(crate) mod tests;