liminal-server 0.25.4

//! The live Lys question over HTTP (ACCESS-002 R1, ACCESS-003 R1).
//!
//! `lys-pass`'s batch route is asynchronous; the connection's decision path
//! is not. Each ask runs its future to completion on a scoped thread against
//! the server's tokio runtime, so a connection process never blocks the
//! runtime's own workers and never nests a runtime. Deliberate asks are rare
//! by definition (opening a link, deploying a service, an administrator's
//! act), so a thread per ask is the honest cost; nothing is cached, and no
//! failure falls back to the hot seam. A held answer is recorded as a draft
//! with the holder's own pass before it is reported, so a held open is a
//! draft Lys retains, never a promise liminal invented.

use super::GrantMode;
use super::lys_pass_adapter::{LiveAnswer, LiveLysAsk};
use lys_pass::deliberate::CheckAnswer;
use lys_pass::drafts::{DraftRequest, request_digest};
use lys_pass::{Client, Error, Mode, Target};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, Weak};
use std::time::{Duration, Instant};

static ASK_METRICS: Mutex<Vec<Weak<AskMetrics>>> = Mutex::new(Vec::new());

#[derive(Debug, Default)]
struct AskMetrics {
    count: AtomicU64,
    maximum_ns: AtomicU64,
}

/// Completed requests across the live Lys clients in this process.
#[derive(Clone, Copy, Debug, Default, serde::Serialize)]
pub(crate) struct AskStatistics {
    pub count: u64,
    pub max_elapsed_ms: f64,
}

impl AskMetrics {
    fn record(&self, elapsed: Duration) {
        let nanos = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX);
        self.maximum_ns.fetch_max(nanos, Ordering::Relaxed);
        self.count.fetch_add(1, Ordering::Release);
    }

    fn snapshot(&self) -> AskStatistics {
        AskStatistics {
            count: self.count.load(Ordering::Acquire),
            max_elapsed_ms: Duration::from_nanos(self.maximum_ns.load(Ordering::Relaxed))
                .as_secs_f64()
                * 1000.0,
        }
    }
}

/// Reads the same per-client counters the request path updates, without a task or timer.
pub(crate) fn ask_statistics() -> Result<AskStatistics, Error> {
    let mut clients = ASK_METRICS
        .lock()
        .map_err(|_| Error::CannotAsk("Lys ask metrics registry poisoned"))?;
    let mut total = AskStatistics::default();
    clients.retain(|client| {
        let Some(client) = client.upgrade() else {
            return false;
        };
        let snapshot = client.snapshot();
        total.count = total.count.saturating_add(snapshot.count);
        total.max_elapsed_ms = total.max_elapsed_ms.max(snapshot.max_elapsed_ms);
        true
    });
    drop(clients);
    Ok(total)
}

#[path = "drafts.rs"]
pub mod drafts;

/// Exact prepared operation; the target owner retains these immutable bytes.
pub struct PreparedAct<'a> {
    /// The holder proved by the admitted pass, never taken from a JSON body.
    pub holder: &'a str,
    /// The authoritative resource and deliberate action.
    pub target: &'a Target,
    /// Stable id of this prepared operation, reused only for the same bytes.
    pub operation: &'a str,
    /// Exact words presented to the approver, unchanged.
    pub words: &'a str,
    /// The server's preparation instant in epoch milliseconds.
    pub at_ms: u64,
}

/// The target registry keeps preparation before the remote draft is created,
///
/// then binds its returned id and waiting state durably. A failed handoff is
/// named; it never turns the held decision into permission to execute.
pub trait PreparedDraftKeeper: Send + Sync {
    /// Retains the immutable prepared act under its operation and digest.
    ///
    /// # Errors
    /// A conflicting operation, configured bound or store failure.
    fn prepare(&self, holder: &str, request: &DraftRequest, at_ms: u64) -> Result<(), Error>;
    /// Binds the retained draft and writes waiting through the agent state writer.
    ///
    /// # Errors
    /// Binding or durable waiting-state failure.
    fn held(&self, holder: &str, id: &str, request: &DraftRequest, at_ms: u64)
    -> Result<(), Error>;
}

/// Prepared answers carry held words as data and exclude them from diagnostics.
pub struct PreparedAnswer {
    /// The actual Lys decision, still held when approval is required.
    pub decision: LiveAnswer,
    /// Exact words of a held act, absent for outright or refused decisions.
    pub words: Option<String>,
}

impl std::fmt::Debug for PreparedAnswer {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PreparedAnswer")
            .finish_non_exhaustive()
    }
}
/// `lys-pass`'s client behind the [`LiveLysAsk`] seam.
pub struct HttpLysAsk {
    client: Client,
    metrics: Arc<AskMetrics>,
    handle: tokio::runtime::Handle,
    /// This liminal's Lys app credential, which the batch check is asked
    /// with when present; absent, the holder's pass is the bearer.
    asker: Option<String>,
    keeper: Option<Arc<dyn PreparedDraftKeeper>>,
}

/// The Lys origin the lys-pass client takes (from 0.3.3 it mounts `/api`
/// itself): a configured API mount (`api_url`, `.../api/`) names its origin.
///
/// # Errors
/// The value is not a URL.
pub fn lys_origin(url: &str) -> Result<reqwest::Url, Error> {
    let mut url: reqwest::Url = url
        .parse()
        .map_err(|_| Error::Invalid("issuer is not a URL"))?;
    if matches!(url.path(), "/api" | "/api/") {
        url.set_path("/");
    }
    Ok(url)
}

impl std::fmt::Debug for HttpLysAsk {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.debug_struct("HttpLysAsk").finish_non_exhaustive()
    }
}

impl HttpLysAsk {
    /// Asks `issuer` through a fresh HTTP client driven by `handle`'s runtime.
    ///
    /// # Errors
    /// The issuer is not an HTTP origin.
    pub fn new(issuer: &str, handle: tokio::runtime::Handle) -> Result<Self, Error> {
        let issuer = lys_origin(issuer)?;
        let metrics = Arc::new(AskMetrics::default());
        {
            let mut clients = ASK_METRICS
                .lock()
                .map_err(|_| Error::CannotAsk("Lys ask metrics registry poisoned"))?;
            clients.retain(|client| client.strong_count() > 0);
            clients.push(Arc::downgrade(&metrics));
        }
        Ok(Self {
            client: Client::new(
                reqwest::Client::builder()
                    .no_proxy()
                    .redirect(reqwest::redirect::Policy::none())
                    .tcp_keepalive(Duration::from_secs(30))
                    .pool_idle_timeout(None)
                    .build()
                    .map_err(|error| Error::Transport(Box::new(error)))?,
                issuer,
            )?,
            handle,
            metrics,
            asker: None,
            keeper: None,
        })
    }

    /// Asks the batch check as this liminal's app, `credential`, about the
    /// holder, instead of with the holder's own pass.
    #[must_use]
    pub fn asking_as(mut self, credential: String) -> Self {
        self.asker = Some(credential);
        self
    }

    /// Installs the target registry preparation and waiting writer.
    #[must_use]
    pub fn with_drafts(mut self, keeper: Arc<dyn PreparedDraftKeeper>) -> Self {
        self.keeper = Some(keeper);
        self
    }

    #[cfg(test)]
    fn statistics(&self) -> AskStatistics {
        self.metrics.snapshot()
    }

    /// Runs one ask to completion on its own thread against the runtime.
    fn block<T, F>(&self, route: &'static str, future: F) -> Result<T, Error>
    where
        T: Send,
        F: std::future::Future<Output = Result<T, Error>> + Send,
    {
        let started = Instant::now();
        let answer = drafts::block(&self.handle, future);
        let elapsed = started.elapsed();
        self.metrics.record(elapsed);
        tracing::debug!(target: "liminal::lys_ask", route, ok = answer.is_ok(),
            elapsed_ms = elapsed.as_secs_f64() * 1000.0, "Lys ask completed");
        answer
    }

    fn draft(
        &self,
        pass: &str,
        act: &PreparedAct<'_>,
        grant: &str,
        judgment: &CheckAnswer,
    ) -> Result<String, Error> {
        let draft = DraftRequest {
            operation: act.operation.to_owned(),
            grant: grant.to_owned(),
            target: Target::new(&act.target.kind, &act.target.id, &act.target.action)?,
            request_digest: request_digest(act.words),
            words: act.words.to_owned(),
        };
        if let Some(keeper) = &self.keeper {
            keeper.prepare(act.holder, &draft, act.at_ms)?;
        }
        let created = self.block(
            "/api/product-drafts",
            self.client.create_draft(pass, &draft, judgment),
        )?;
        if let Some(keeper) = &self.keeper {
            keeper.held(act.holder, &created.draft, &draft, act.at_ms)?;
        }
        Ok(created.draft)
    }
    /// Asks about a complete prepared act, preserving the exact approver words.
    ///
    /// # Errors
    /// Missing keeper, invalid or oversized input, or Lys's live refusal.
    pub fn ask_prepared(
        &self,
        pass: &str,
        act: &PreparedAct<'_>,
        limits: drafts::DraftLimits,
    ) -> Result<PreparedAnswer, Error> {
        if self.keeper.is_none() {
            return Err(Error::CannotAsk("draft preparation writer is missing"));
        }
        // The pass is a credential, bounded at admission (credential_bytes), not draft metadata.
        if pass.is_empty()
            || act.words.is_empty()
            || act.words.len() > limits.words_bytes.get()
            || [
                act.holder,
                act.operation,
                &act.target.kind,
                &act.target.id,
                &act.target.action,
            ]
            .iter()
            .any(|value| {
                value.is_empty()
                    || value.len() > limits.metadata_bytes.get()
                    || value.chars().any(char::is_control)
            })
        {
            return Err(Error::Invalid("prepared draft bound exceeded"));
        }
        let decision = self.prepared_decision(pass, act)?;
        let words = matches!(decision, LiveAnswer::Held { .. }).then(|| act.words.to_owned());
        Ok(PreparedAnswer { decision, words })
    }

    fn prepared_decision(&self, pass: &str, act: &PreparedAct<'_>) -> Result<LiveAnswer, Error> {
        let answer = self.block(
            "/api/grants/check/batch",
            self.client.check_batch(
                self.asker.as_deref().unwrap_or(pass),
                act.holder,
                std::slice::from_ref(act.target),
                None,
            ),
        )?;
        if answer.degraded.is_some() {
            return Err(Error::CannotAsk("live permission projection degraded"));
        }
        let Some(result) = answer.results.into_iter().next() else {
            return Err(Error::CannotAsk("the live answer carried no result"));
        };
        if result.degraded.is_some() {
            return Err(Error::CannotAsk("live permission projection degraded"));
        }
        match (
            result.allowed,
            result.mode.as_ref(),
            result.grant.as_deref(),
            result.refusal.as_deref(),
        ) {
            (true, Some(Mode::Outright), Some(grant), None) => Ok(LiveAnswer::Outright {
                grant: grant.to_owned(),
            }),
            (false, Some(Mode::ByDraft), Some(grant), None) => Ok(LiveAnswer::Held {
                mode: GrantMode::ByDraft,
                grant: grant.to_owned(),
                draft: self.draft(pass, act, grant, &result)?,
            }),
            (false, Some(Mode::ByTwo), Some(grant), None) => Ok(LiveAnswer::Held {
                mode: GrantMode::ByTwo,
                grant: grant.to_owned(),
                draft: self.draft(pass, act, grant, &result)?,
            }),
            (false, None, None, Some(refusal)) => Ok(LiveAnswer::Refused {
                refusal: refusal.to_owned(),
            }),
            _ => Err(Error::CannotAsk("inconsistent live permission answer")),
        }
    }
}

impl LiveLysAsk for HttpLysAsk {
    fn ask(
        &self,
        pass: &str,
        holder: &str,
        target: &Target,
        now: u64,
    ) -> Result<LiveAnswer, Error> {
        let words = format!(
            "{holder} asks {} on {}:{}",
            target.action, target.kind, target.id
        );
        let operation = format!(
            "liminal:{holder}:{}:{}:{}",
            target.kind, target.id, target.action
        );
        let at_ms = now
            .checked_mul(1000)
            .ok_or(Error::Invalid("draft instant overflow"))?;
        self.prepared_decision(
            pass,
            &PreparedAct {
                holder,
                target,
                operation: &operation,
                words: &words,
                at_ms,
            },
        )
    }
}

#[path = "live/drafts_join.rs"]
mod drafts_join;
pub use drafts_join::PreparedOwner;

#[cfg(test)]
#[path = "live/timing_tests.rs"]
mod timing_tests;