haematite-cli 0.21.3

//! A served connection's right, decided from its Lys pass or by asking Lys
//! (ACCESS-001 R2, PERMISSIONS-ACROSS-THE-STACK D2, D3, D4 and D9).
//!
//! A hot route is decided from the verified pass alone and never asks Lys. It
//! accepts only an outright grant: a grant held by draft or by two never
//! applies to a hot route, so it is refused in the one refusal shape. A
//! deliberate route asks Lys live with the pass as the caller. When Lys cannot
//! answer, the act is refused "Lys could not be asked" and the pass is never
//! consulted in its place.
//!
//! Lys expands reach and roles when it issues a pass: a database grant arrives
//! as explicit rights on the collections, branches and backups placed under
//! it. So every question here names one exact resource, and nothing is
//! recomputed locally.

use std::collections::BTreeMap;
use std::sync::{Arc, Condvar, Mutex, OnceLock, PoisonError, Weak};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use lys_pass::deliberate::CheckAnswer;
use lys_pass::drafts::{DraftRequest, RunError, RunReport, request_digest};
use lys_pass::{Decision, Error, KeySet, Mode, Refusal, Target, VerifiedPass};
use serde_json::{Value, json};

use crate::serve::authorize::subjects;
use crate::serve::drafts::Drafts;
use crate::serve::protocol::Operation;
use crate::serve::route_matrix::{Class, Route};
use crate::serve::service::Service;
use crate::serve::storage::Failure;

/// The audience every haematite pass is issued for.
pub const AUDIENCE: &str = "haematite";

/// The name a route answers when the connection presented no pass.
pub const PASS_MISSING: &str = "pass_missing";

/// Every decided act is refused by this name until Lys's published keys have
/// been read: the service listens at once and never decides without them.
pub const LYS_KEYS_PENDING: &str = "lys_keys_pending";

/// The first wait before Lys's keys are asked again, doubled after every
/// failure up to [`KEYS_RETRY_MOST`].
const KEYS_RETRY_FIRST: Duration = Duration::from_millis(250);

/// The longest wait between two asks for Lys's keys.
const KEYS_RETRY_MOST: Duration = Duration::from_secs(30);

/// What a served connection trusts: Lys's published keys, its issuer and this
/// product's audience.
#[derive(Clone, Copy)]
pub struct Trust<'a> {
    /// The issuer's published verification keys.
    pub keys: &'a KeySet,
    /// The trusted issuer origin.
    pub issuer: &'a str,
    /// The audience a pass must be issued for.
    pub audience: &'a str,
}

/// Asks Lys live whether the pass's holder may act on one target, and
/// records an act held for approval as a Lys draft.
pub trait LiveCheck {
    /// One live answer for `target`, with `token` as the caller.
    ///
    /// # Errors
    ///
    /// Any [`Error`] means Lys could not be asked or answered outside its
    /// contract; the act is then refused and the pass is never consulted.
    fn ask(&self, token: &str, holder: &str, target: &Target) -> Result<CheckAnswer, Error>;

    /// Record `request` as a draft with `token` as the caller, under the
    /// held `judgment` Lys answered for it, answering the draft's id.
    ///
    /// # Errors
    ///
    /// Any [`Error`] means the draft was not recorded; the act is refused
    /// and nothing runs in its place.
    fn draft(
        &self,
        token: &str,
        request: &DraftRequest,
        judgment: &CheckAnswer,
    ) -> Result<String, Error>;

    /// Pull one page of approved drafts with `connector` as the caller,
    /// after `after`, apply each exactly once to `drafts` and acknowledge it.
    /// A live check that cannot reach Lys runs nothing.
    ///
    /// # Errors
    ///
    /// Where the run stopped; `Pull` with `lys_could_not_be_asked` here.
    fn run_drafts(
        &self,
        _connector: &str,
        _drafts: &mut Drafts<'_>,
        _after: Option<&str>,
    ) -> Result<RunReport, Box<RunError<Failure>>> {
        Err(Box::new(RunError::Pull(Error::CannotAsk(
            "this live check runs no drafts",
        ))))
    }
}

/// A live check that can never reach Lys, so every deliberate route is refused
/// "Lys could not be asked" (D10: availability is stated, never a fallback).
pub struct Unreachable;

impl LiveCheck for Unreachable {
    fn ask(&self, _token: &str, _holder: &str, _target: &Target) -> Result<CheckAnswer, Error> {
        Err(Error::CannotAsk("no live check is configured"))
    }

    fn draft(
        &self,
        _token: &str,
        _request: &DraftRequest,
        _judgment: &CheckAnswer,
    ) -> Result<String, Error> {
        Err(Error::CannotAsk("no live check is configured"))
    }
}

/// A route's answer for one act.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Verdict {
    /// The act happens, receipted with this grant (D11).
    Allowed {
        /// The granting receipt recorded with the act.
        grant: String,
    },
    /// A deliberate act the holder may only do by draft or by two (D4): it is
    /// written as a Lys draft and done after approval.
    Held {
        /// The grant the draft is held under.
        grant: String,
        /// The approval the draft needs.
        mode: Mode,
        /// Lys's live answer, which the draft is recorded under.
        judgment: Box<CheckAnswer>,
    },
    /// No grant permits the act: the one refusal shape (D9).
    Refused(Refusal),
    /// The pass or Lys was refused by its stable name.
    Named(String),
}

impl Verdict {
    fn named(error: &Error) -> Self {
        Self::Named(error.name().to_owned())
    }

    fn refused(trust: Trust<'_>, target: &Target) -> Self {
        match Refusal::new(
            trust.audience,
            &target.kind,
            &target.id,
            &target.action,
            trust.issuer,
        ) {
            Ok(refusal) => Self::Refused(refusal),
            Err(error) => Self::named(&error),
        }
    }
}

/// The Lys resource a route acts on in the served database `database`.
///
/// A database is named by its id; a collection or a branch is placed under
/// its database and named `{database}.{name}`, the id its placement carries
/// in Lys. Lys takes no `/` in an id, and a database id holds no `.`, so the
/// first `.` always ends the database's part and no two names meet.
///
/// # Errors
///
/// Returns [`Error`] when an id is empty or holds a control character.
pub fn target(route: &Route, database: &str, name: Option<&str>) -> Result<Target, Error> {
    let kind = format!("{AUDIENCE}.{}", route.kind);
    match (route.kind, name) {
        ("database", _) => Target::new(&kind, database, route.action),
        (_, Some(name)) => Target::new(&kind, &format!("{database}.{name}"), route.action),
        (_, None) => Err(Error::Invalid("a placed resource needs its name")),
    }
}

/// Decide one act on `target` under `class`, from the pass `token`.
///
/// Every refusal is returned as a [`Verdict`]; nothing here fails open.
#[must_use]
pub fn decide(
    trust: Trust<'_>,
    token: Option<&str>,
    target: &Target,
    class: Class,
    live: &dyn LiveCheck,
    now: u64,
) -> Verdict {
    let Some(token) = token else {
        return Verdict::Named(PASS_MISSING.to_owned());
    };
    let pass = match VerifiedPass::verify(token, trust.keys, trust.issuer, trust.audience, now) {
        Ok(pass) => pass,
        Err(error) => return Verdict::named(&error),
    };
    match class {
        Class::Hot => match pass.evaluate(&target.kind, &target.id, &target.action, now) {
            Ok(Decision::Allowed { grant }) => Verdict::Allowed {
                grant: grant.to_owned(),
            },
            Ok(Decision::Held { .. } | Decision::Refused) => Verdict::refused(trust, target),
            Err(error) => Verdict::named(&error),
        },
        Class::Deliberate => match live.ask(token, &pass.claims().holder.id, target) {
            Ok(answer) => deliberate_verdict(trust, target, &answer),
            Err(error) => Verdict::named(&error),
        },
    }
}

fn deliberate_verdict(trust: Trust<'_>, target: &Target, answer: &CheckAnswer) -> Verdict {
    match (answer.allowed, answer.grant.as_deref(), answer.mode) {
        (true, Some(grant), Some(Mode::Outright)) => Verdict::Allowed {
            grant: grant.to_owned(),
        },
        (false, Some(grant), Some(mode @ (Mode::ByDraft | Mode::ByTwo))) => Verdict::Held {
            grant: grant.to_owned(),
            mode,
            judgment: Box::new(answer.clone()),
        },
        (false, None, None) => Verdict::refused(trust, target),
        _ => Verdict::Named(
            Error::CannotAsk("inconsistent live permission answer")
                .name()
                .to_owned(),
        ),
    }
}

/// A served haematite's Lys gate: the keys, issuer and database id it
/// decides every request against, and the live check deliberate routes ask.
pub struct Gate {
    /// Lys's published keys, empty until they have been read.
    keys: Arc<OnceLock<Mutex<Arc<KeySet>>>>,
    refresh: Option<Mutex<KeyRefresh>>,
    issuer: String,
    database: String,
    live: Arc<dyn LiveCheck + Send + Sync>,
    /// What this service made: a pass issued before it cannot carry its
    /// rights, so a hot refusal on it is asked of Lys live.
    made: Arc<Made>,
}

type FetchKeys = Arc<dyn Fn() -> Result<KeySet, Error> + Send + Sync>;

struct KeyRefresh {
    fetch: FetchKeys,
    last: Option<Instant>,
    interval: Duration,
}

/// How long a made resource is remembered: longer than any pass Lys issues
/// lives, so a pass issued before the resource was made has expired before
/// the resource is forgotten.
const PLACED_KEPT_SECS: u64 = 24 * 60 * 60;

/// How long a refused act waits to be told that what it names has been
/// placed in Lys, before Lys is asked: the placement's own answer bound
/// (serve/placement.rs), so a placement that never answers ends the wait.
const PLACEMENT_WAIT: Duration = Duration::from_secs(10);

/// Each collection and branch this service made, by its Lys kind and name.
///
/// Each is kept with when its write committed, in Unix seconds, and whether
/// its placement in Lys has been answered. Placing runs on its own thread after the commit
/// (serve/placement.rs), so Lys, asked about what was just made, would not
/// yet hold it; a gate asking about it is told when the placement lands.
#[derive(Debug, Default)]
pub struct Made {
    state: Mutex<MadeState>,
    landed: Condvar,
}

#[derive(Debug, Default)]
struct MadeState {
    /// By Lys kind and name: when it committed, and whether it has landed.
    made: BTreeMap<(String, String), (u64, bool)>,
    /// Waits begun for a placement to land, ever: a count that only grows.
    waits: u64,
}

impl Made {
    fn committed(&self, kind: &str, name: &str, now: u64, placing: bool) {
        let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
        state
            .made
            .retain(|_, (at, _)| now.saturating_sub(*at) < PLACED_KEPT_SECS);
        state
            .made
            .insert((kind.to_owned(), name.to_owned()), (now, !placing));
    }

    /// The placement of `name` (Lys kind `kind`) was answered: placed,
    /// already placed, or failed. A gate waiting on it asks Lys now.
    pub fn landed(&self, kind: &str, name: &str) {
        if let Some((_, landed)) = self
            .state
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .made
            .get_mut(&(kind.to_owned(), name.to_owned()))
        {
            *landed = true;
        }
        self.landed.notify_all();
    }

    /// When `name` was made, once its placement has been answered or
    /// `within` has passed; `None` when this service did not make it.
    fn made_at(&self, kind: &str, name: &str, within: Duration) -> Option<u64> {
        let key = (kind.to_owned(), name.to_owned());
        {
            let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
            state.waits = state.waits.saturating_add(1);
        }
        self.landed.notify_all();
        // a landing between the two locks is seen by the wait's first check
        let (ended, _) = self
            .landed
            .wait_timeout_while(
                self.state.lock().unwrap_or_else(PoisonError::into_inner),
                within,
                |state| state.made.get(&key).is_some_and(|(_, landed)| !landed),
            )
            .unwrap_or_else(PoisonError::into_inner);
        let made = ended.made.get(&key).map(|(at, _)| *at);
        drop(ended);
        made
    }
}

#[cfg(test)]
impl Made {
    /// Wait until an act has begun waiting for a placement to land.
    pub(super) fn await_waiting(&self) {
        drop(
            self.landed
                .wait_while(
                    self.state.lock().unwrap_or_else(PoisonError::into_inner),
                    |state| state.waits == 0,
                )
                .unwrap_or_else(PoisonError::into_inner),
        );
    }
}

impl Gate {
    /// A gate for the served database `database`, trusting `issuer` and the
    /// keys it publishes as `keys_json`.
    ///
    /// # Errors
    ///
    /// Returns [`Error`] when the published keyset does not validate whole.
    pub fn new(
        keys_json: &str,
        issuer: &str,
        database: &str,
        live: Arc<dyn LiveCheck + Send + Sync>,
    ) -> Result<Self, Error> {
        Ok(Self::with_keys(
            KeySet::from_json(keys_json)?,
            issuer,
            database,
            live,
        ))
    }

    /// A gate trusting `keys`, as read from `issuer` (`/oauth/jwks`).
    #[must_use]
    pub fn with_keys(
        keys: KeySet,
        issuer: &str,
        database: &str,
        live: Arc<dyn LiveCheck + Send + Sync>,
    ) -> Self {
        Self {
            keys: Arc::new(OnceLock::from(Mutex::new(Arc::new(keys)))),
            refresh: None,
            issuer: issuer.to_owned(),
            database: database.to_owned(),
            live,
            made: Arc::new(Made::default()),
        }
    }

    /// A gate for `database` trusting `issuer`, listening before its keys are
    /// read: `fetch` asks Lys for them on a thread of its own, again after
    /// every failure, waiting from [`KEYS_RETRY_FIRST`] doubled up to
    /// [`KEYS_RETRY_MOST`], until they are read or the gate is dropped. Until
    /// then every decided act is refused as [`LYS_KEYS_PENDING`].
    ///
    /// # Errors
    ///
    /// Returns the operating system's error when the thread cannot start.
    pub fn fetching(
        fetch: impl Fn() -> Result<KeySet, Error> + Send + Sync + 'static,
        issuer: &str,
        database: &str,
        live: Arc<dyn LiveCheck + Send + Sync>,
    ) -> std::io::Result<Self> {
        // Unknown ids can trigger one refresh per configured interval, including failed fetches.
        let interval = crate::door::config::positive_env("HAEMATITE_LYS_KEYS_REFRESH_SECONDS", 30)
            .map_err(std::io::Error::other)?;
        let fetch: FetchKeys = Arc::new(fetch);
        let retry = Arc::clone(&fetch);
        let keys = Arc::new(OnceLock::new());
        let landing = Arc::downgrade(&keys);
        let from = issuer.to_owned();
        std::thread::Builder::new()
            .name("haematite-lys-keys".to_owned())
            .spawn(move || read_keys(retry.as_ref(), &landing, &from))?;
        Ok(Self {
            keys,
            refresh: Some(Mutex::new(KeyRefresh {
                fetch,
                last: None,
                interval: Duration::from_secs(interval.get()),
            })),
            issuer: issuer.to_owned(),
            database: database.to_owned(),
            live,
            made: Arc::new(Made::default()),
        })
    }

    /// What this gate knows was made, for the placer to say when each lands.
    #[must_use]
    pub fn made(&self) -> Arc<Made> {
        Arc::clone(&self.made)
    }

    /// A collection or branch named `name` (its Lys kind `kind`) was made by
    /// a write that committed at `now`; `placing` when a placer will place it
    /// and say so. What was made a day before is let go.
    pub fn placed(&self, kind: &str, name: &str, now: u64, placing: bool) {
        self.made.committed(kind, name, now, placing);
    }

    /// For a hot act the pass refused: when `target` was made at or after
    /// the pass was issued, Lys is asked live once, and an outright answer
    /// is the grant. Anything else leaves the refusal standing.
    fn placed_after_pass(&self, token: Option<&str>, target: &Target, now: u64) -> Option<String> {
        let token = token?;
        let keys = self.keys_for(Some(token), now).ok()?;
        let trust = self.trust(&keys);
        let name = target.id.strip_prefix(&format!("{}.", self.database))?;
        let made = self.made.made_at(&target.kind, name, PLACEMENT_WAIT)?;
        let pass = VerifiedPass::verify(token, trust.keys, trust.issuer, AUDIENCE, now).ok()?;
        if made < pass.claims().iat {
            return None;
        }
        let answer = self
            .live
            .ask(token, &pass.claims().holder.id, target)
            .ok()?;
        match deliberate_verdict(trust, target, &answer) {
            Verdict::Allowed { grant } => Some(grant),
            _ => None,
        }
    }

    /// The pass's outright grant of `action` on this gate's database.
    fn database_right(&self, token: Option<&str>, action: &str, now: u64) -> Option<String> {
        let target = Target::new(&format!("{AUDIENCE}.database"), &self.database, action).ok()?;
        let keys = self.keys_for(token, now).ok()?;
        match decide(
            self.trust(&keys),
            token,
            &target,
            Class::Hot,
            self.live.as_ref(),
            now,
        ) {
            Verdict::Allowed { grant } => Some(grant),
            _ => None,
        }
    }

    /// What this gate trusts, refused as [`LYS_KEYS_PENDING`] until Lys's
    /// keys have been read.
    fn snapshot(&self) -> Result<Arc<KeySet>, Failure> {
        let held = self
            .keys
            .get()
            .ok_or_else(|| named_failure(LYS_KEYS_PENDING))?;
        held.lock()
            .map(|keys| Arc::clone(&keys))
            .map_err(|_| Failure::new("lys_keys_unavailable", "Lys key state was poisoned"))
    }

    fn keys_for(&self, token: Option<&str>, now: u64) -> Result<Arc<KeySet>, Failure> {
        let keys = self.snapshot()?;
        let Some(token) = token else {
            return Ok(keys);
        };
        let unknown = |keys: &KeySet| {
            VerifiedPass::verify(token, keys, &self.issuer, AUDIENCE, now)
                .is_err_and(|error| error.name() == "unpublished_key")
        };
        if !unknown(&keys) {
            return Ok(keys);
        }
        let Some(refresh) = &self.refresh else {
            return Ok(keys);
        };
        let mut refresh = refresh
            .lock()
            .map_err(|_| Failure::new("lys_keys_unavailable", "Lys key refresh was poisoned"))?;
        let current = self.snapshot()?;
        if !unknown(&current)
            || refresh
                .last
                .is_some_and(|last| last.elapsed() < refresh.interval)
        {
            return Ok(current);
        }
        refresh.last = Some(Instant::now());
        let fresh = (refresh.fetch)().map_err(|error| {
            eprintln!("haem/1 lys key refresh refused: {}", error.name());
            named_failure(error.name())
        })?;
        let fresh = Arc::new(fresh);
        let held = self
            .keys
            .get()
            .ok_or_else(|| named_failure(LYS_KEYS_PENDING))?;
        *held
            .lock()
            .map_err(|_| Failure::new("lys_keys_unavailable", "Lys key state was poisoned"))? =
            Arc::clone(&fresh);
        // Keep the refresh serialized until its snapshot is published.
        drop(refresh);
        eprintln!("haem/1 lys keys refreshed");
        Ok(fresh)
    }

    fn trust<'a>(&'a self, keys: &'a KeySet) -> Trust<'a> {
        Trust {
            keys,
            issuer: &self.issuer,
            audience: AUDIENCE,
        }
    }

    /// Verify a presented pass and answer whose it is and when it expires.
    ///
    /// # Errors
    ///
    /// Returns the pass's refusal by its stable name.
    pub fn present(&self, token: &str, now: u64) -> Result<Value, Failure> {
        let keys = self.keys_for(Some(token), now)?;
        let trust = self.trust(&keys);
        let pass = VerifiedPass::verify(token, trust.keys, trust.issuer, AUDIENCE, now)
            .map_err(|error| named_failure(error.name()))?;
        let claims = pass.claims();
        Ok(json!({"holder": claims.holder.id, "expires": claims.exp}))
    }

    /// Decide every subject of `operation` from the connection's pass `token`,
    /// answering the grants that permit it, one per subject (D11). `words`
    /// are the request's exact text, which a held change is drafted with.
    ///
    /// # Errors
    ///
    /// Returns the first refusal: `permission_refused` carrying the one
    /// refusal shape, `held_for_approval` for a grant held by draft or by two,
    /// or the pass's or Lys's stable name.
    pub fn authorize(
        &self,
        token: Option<&str>,
        operation: &Operation,
        words: Option<&str>,
        now: u64,
    ) -> Result<Vec<String>, Failure> {
        self.authorize_with(token, operation, words, now, &|_, _| false)
    }

    /// [`Gate::authorize`], where `absent(kind, name)` answers whether the
    /// store holds no such branch, or no such collection on the change's
    /// branch. Lys places only what exists, so a pass carries no right on what
    /// does not: a hot act refused on it is decided by the same right on the
    /// database, the right it would inherit, and the store then answers
    /// `branch_missing` or `collection_missing`.
    ///
    /// # Errors
    ///
    /// As [`Gate::authorize`].
    pub fn authorize_with(
        &self,
        token: Option<&str>,
        operation: &Operation,
        words: Option<&str>,
        now: u64,
        absent: &dyn Fn(&str, &str) -> bool,
    ) -> Result<Vec<String>, Failure> {
        let subjects = subjects(operation)?;
        if subjects.is_empty() {
            return Ok(Vec::new());
        }
        let keys = self.keys_for(token, now)?;
        let alone = subjects.len() == 1;
        let mut grants = Vec::new();
        for subject in subjects {
            let target = target(subject.route, &self.database, subject.name.as_deref())
                .map_err(|error| named_failure(error.name()))?;
            match decide(
                self.trust(&keys),
                token,
                &target,
                subject.route.class,
                self.live.as_ref(),
                now,
            ) {
                Verdict::Allowed { grant } => grants.push(grant),
                Verdict::Held { judgment, .. } if alone => {
                    return Err(self.hold(token, operation, words, &target, &judgment));
                }
                Verdict::Held { grant, .. } => {
                    return Err(Failure::new(
                        "held_for_approval",
                        format!(
                            "held under grant {grant}; an act on more than one resource is never drafted"
                        ),
                    ));
                }
                Verdict::Refused(_)
                    if subject.route.class == Class::Hot
                        && let Some(grant) = self.placed_after_pass(token, &target, now) =>
                {
                    grants.push(grant);
                }
                Verdict::Refused(_)
                    if subject.route.class == Class::Hot
                        && subject
                            .name
                            .as_deref()
                            .is_some_and(|name| absent(subject.route.kind, name))
                        && let Some(grant) =
                            self.database_right(token, subject.route.action, now) =>
                {
                    grants.push(grant);
                }
                Verdict::Refused(refusal) => {
                    let shape = serde_json::to_string(&refusal).map_err(Failure::storage)?;
                    return Err(Failure::new("permission_refused", shape));
                }
                Verdict::Named(name) => return Err(named_failure(&name)),
            }
        }
        Ok(grants)
    }

    /// Record a held change as a Lys draft of its exact `words`, answering
    /// `held_for_approval` with the draft (D4). Only a change is drafted: a
    /// held read has nothing to apply after approval. The draft's operation
    /// id is the change's request id scoped to this database and its branch,
    /// so a retry of the same request finds the same draft.
    fn hold(
        &self,
        token: Option<&str>,
        operation: &Operation,
        words: Option<&str>,
        target: &Target,
        judgment: &CheckAnswer,
    ) -> Failure {
        let Some(token) = token else {
            return named_failure(PASS_MISSING);
        };
        // A held verdict is only made from an answer carrying both.
        let (Some(grant), Some(mode)) = (judgment.grant.as_deref(), judgment.mode) else {
            return named_failure(Error::CannotAsk("inconsistent live permission answer").name());
        };
        let (Operation::Mutate(mutation), Some(words)) = (operation, words) else {
            return Failure::new(
                "held_for_approval",
                format!("held under grant {grant}; only a change is drafted for approval"),
            );
        };
        let request = DraftRequest {
            operation: format!(
                "{AUDIENCE}:{}:{}:{}",
                self.database, mutation.branch, mutation.request_id
            ),
            grant: grant.to_owned(),
            target: target.clone(),
            request_digest: request_digest(words),
            words: words.to_owned(),
        };
        match self.live.draft(token, &request, judgment) {
            Ok(draft) => Failure::new(
                "held_for_approval",
                json!({"draft": draft, "grant": grant, "mode": mode}).to_string(),
            ),
            Err(error) => named_failure(error.name()),
        }
    }

    /// Run one page of the drafts Lys approved for this gate's database,
    /// applied to `service`, with the connection's pass `token` as the
    /// connector's. The route was decided before this, as `drafts.run`.
    ///
    /// # Errors
    ///
    /// `pass_missing` without a pass, or where the run stopped, by name: a
    /// change applied but not acknowledged keeps its receipt, and the next
    /// run answers it again without applying it twice.
    pub fn run_drafts(
        &self,
        service: &Service,
        token: Option<&str>,
        after: Option<&str>,
    ) -> Result<Value, Failure> {
        let Some(token) = token else {
            return Err(named_failure(PASS_MISSING));
        };
        let mut drafts = Drafts::new(service, &self.database);
        match self
            .live
            .run_drafts(token, &mut drafts, after)
            .map_err(|error| *error)
        {
            Ok(report) => Ok(json!({"acknowledged": report.acknowledged, "next": report.next})),
            Err(RunError::Pull(error)) => Err(named_failure(error.name())),
            Err(RunError::Execution { draft, source }) => Err(Failure::new(
                "draft_execution_failed",
                format!("draft {draft} was not applied: {source}"),
            )),
            Err(RunError::Record { draft, source, .. }) => Err(Failure::new(
                "draft_acknowledgment_failed",
                format!(
                    "draft {draft} was applied but Lys did not take its outcome ({}); \
                     the next run answers it again from its receipt",
                    source.name()
                ),
            )),
        }
    }
}

/// The current time in seconds since the epoch, as passes count it.
#[must_use]
pub fn now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |elapsed| elapsed.as_secs())
}

/// Ask `fetch` for Lys's keys until they land in `landing`, or until the
/// gate that holds it is dropped.
fn read_keys(
    fetch: &dyn Fn() -> Result<KeySet, Error>,
    landing: &Weak<OnceLock<Mutex<Arc<KeySet>>>>,
    issuer: &str,
) {
    let mut wait = KEYS_RETRY_FIRST;
    loop {
        if landing.strong_count() == 0 {
            return;
        }
        match fetch() {
            Ok(keys) => {
                if landing
                    .upgrade()
                    .is_some_and(|held| held.set(Mutex::new(Arc::new(keys))).is_ok())
                {
                    eprintln!("haem/1 lys keys read from {issuer}");
                }
                return;
            }
            Err(error) => {
                eprintln!(
                    "haem/1 lys keys pending: {} from {issuer}; asking again in {} ms",
                    error.name(),
                    wait.as_millis()
                );
                std::thread::sleep(wait);
                wait = wait.saturating_mul(2).min(KEYS_RETRY_MOST);
            }
        }
    }
}

fn named_failure(name: &str) -> Failure {
    let words = match name {
        "lys_could_not_be_asked" => "Lys could not be asked",
        PASS_MISSING => "present a Lys pass with session.pass first",
        LYS_KEYS_PENDING => "Lys's published keys have not been read yet",
        _ => "the Lys pass was refused",
    };
    Failure::new(name, words)
}

#[cfg(test)]
#[path = "pass_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "pass_draft_tests.rs"]
mod draft_tests;

#[cfg(test)]
#[path = "pass_collection_tests.rs"]
mod collection_tests;

#[cfg(test)]
#[path = "pass_merge_tests.rs"]
mod merge_tests;

#[cfg(test)]
#[path = "pass_keys_tests.rs"]
mod keys_tests;

#[cfg(test)]
#[path = "pass_rotation_tests.rs"]
mod rotation_tests;