liminal-server 0.23.1

//! 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)
}

/// `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>,
}

/// 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()
                    .tcp_keepalive(Duration::from_secs(30))
                    .pool_idle_timeout(None)
                    .build()
                    .map_err(|error| Error::Transport(Box::new(error)))?,
                issuer,
            )?,
            handle,
            metrics,
            asker: 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
    }

    #[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 handle = &self.handle;
        let answer =
            std::thread::scope(|scope| scope.spawn(move || handle.block_on(future)).join())
                .unwrap_or_else(|_| Err(Error::CannotAsk("the live ask did not complete")));
        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
    }

    /// Records the held act as a draft with the holder's own pass, returning
    /// the draft Lys retained.
    fn draft(
        &self,
        pass: &str,
        holder: &str,
        target: &Target,
        grant: &str,
        judgment: &CheckAnswer,
    ) -> Result<String, Error> {
        let words = format!(
            "{holder} asks {} on {}:{}",
            target.action, target.kind, target.id
        );
        let draft = DraftRequest {
            operation: format!(
                "liminal:{holder}:{}:{}:{}",
                target.kind, target.id, target.action
            ),
            grant: grant.to_owned(),
            target: Target::new(&target.kind, &target.id, &target.action)?,
            request_digest: request_digest(&words),
            words,
        };
        self.block(
            "/api/product-drafts",
            self.client.create_draft(pass, &draft, judgment),
        )
        .map(|created| created.draft)
    }
}

impl LiveLysAsk for HttpLysAsk {
    fn ask(
        &self,
        pass: &str,
        holder: &str,
        target: &Target,
        _now: u64,
    ) -> Result<LiveAnswer, Error> {
        let answer = self.block(
            "/api/grants/check/batch",
            self.client.check_batch(
                self.asker.as_deref().unwrap_or(pass),
                holder,
                std::slice::from_ref(target),
                None,
            ),
        )?;
        let Some(result) = answer.results.into_iter().next() else {
            return Err(Error::CannotAsk("the live answer carried no result"));
        };
        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, holder, target, grant, &result)?,
            }),
            (false, Some(Mode::ByTwo), Some(grant), None) => Ok(LiveAnswer::Held {
                mode: GrantMode::ByTwo,
                grant: grant.to_owned(),
                draft: self.draft(pass, holder, target, grant, &result)?,
            }),
            (false, None, None, Some(refusal)) => Ok(LiveAnswer::Refused {
                refusal: refusal.to_owned(),
            }),
            _ => Err(Error::CannotAsk("inconsistent live permission answer")),
        }
    }
}

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