axond 0.3.25

//! Bounded administrative reads: state, history, audit, convergence.
//!
//! Every projection here is *bounded by its type*, because an administrative read
//! runs against the same control plane a mutation does, and an unbounded one is
//! how a diagnostic query becomes the outage it was meant to diagnose.
//!
//! - **History is a bounded parent walk.** [`ControlPlaneStore`] deliberately
//!   offers no "list revisions" method, so a page is built by following
//!   [`RevisionManifest::parent`] from a starting revision at most
//!   [`HistoryLimit::MAX`] times. There is no query a caller can phrase that
//!   returns more, and retention truncating the chain ends a page instead of
//!   failing it.
//! - **An audit trail is truncated, and says so.** [`AuditPage::truncated`] is
//!   the honest answer to "is this all of it", rather than a response whose size
//!   depends on how eventful a revision was.
//! - **A state read describes resources, never their bodies.** Same rule as the
//!   diff: identity, scope, name, dependencies, and a content checksum.
//! - **Convergence is a projection of what this replica already knows.**
//!   [`ConvergenceResult::of`] reads a [`RevisionReport`], which is cached
//!   replica state; it consults no backend, so it answers during a control-plane
//!   outage — which is precisely when someone asks.
//!
//! Unlike [`crate::status`], these projections *do* name revision ids, mutation
//! ids, and actors: the audience is an administrator of the deployment, and
//! "which revision is desired" is the question. What they still never carry is
//! resource body content, credential material, or raw backend text.
//!
//! [`ControlPlaneStore`]: crate::backends::control_plane::ControlPlaneStore

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde::Serialize;

use super::diff::ScopeView;
use super::error::AdminError;
use crate::convergence::RevisionReport;
use crate::desired_state::{
    Actor, AuditEvent, DesiredState, LoadedRevision, RevisionId, RevisionManifest,
};

/// How many revisions one history page may contain.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HistoryLimit(u32);

impl HistoryLimit {
    pub const MAX: u32 = 100;
    pub const DEFAULT: Self = Self(20);

    /// Refuses zero and anything over [`HistoryLimit::MAX`] rather than clamping:
    /// a caller that asked for 10,000 revisions has a pagination bug, and silently
    /// answering with 100 hides it.
    pub const fn parse(requested: u32) -> Result<Self, AdminError> {
        if requested == 0 || requested > Self::MAX {
            return Err(AdminError::HistoryLimitInvalid { max: Self::MAX });
        }
        Ok(Self(requested))
    }

    pub const fn get(self) -> u32 {
        self.0
    }
}

impl Default for HistoryLimit {
    fn default() -> Self {
        Self::DEFAULT
    }
}

/// A page request: how many revisions, and where to start.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct HistoryRequest {
    pub limit: HistoryLimit,
    /// Start from this revision instead of the newest. The cursor a previous page
    /// returned; a revision is immutable, so paging cannot skip or repeat an
    /// entry the way an offset over mutable rows can.
    pub start: Option<RevisionId>,
}

/// One revision, as history shows it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RevisionRecord {
    pub revision: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent: Option<String>,
    pub created_at_ms: u64,
    /// The mutation that published it, which joins this entry to its audit trail.
    pub mutation: String,
    /// The checksum of the whole desired state, so two deployments can be
    /// compared without shipping either state.
    pub checksum: String,
    pub resources: usize,
    pub blobs: usize,
}

impl RevisionRecord {
    pub fn of(manifest: &RevisionManifest) -> Self {
        Self {
            revision: manifest.id.to_string(),
            parent: manifest.parent.map(|parent| parent.to_string()),
            created_at_ms: millis(manifest.created_at),
            mutation: manifest.mutation.to_string(),
            checksum: manifest.checksum.to_string(),
            resources: manifest.entries.len(),
            blobs: manifest.blobs.len(),
        }
    }
}

/// A bounded page of revisions, newest first.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RevisionPage {
    pub revisions: Vec<RevisionRecord>,
    pub limit: u32,
    /// The revision to pass as `start` for the next page, and `None` when the
    /// walk reached a revision with no parent or one the store no longer retains.
    ///
    /// A cursor is emitted without being loaded, so a page boundary that happens
    /// to fall on the retention edge names a revision that may be pruned before
    /// the caller asks for it. Following it then answers `revision_not_found`
    /// rather than an empty page: a start the caller named is checked, whether it
    /// came from a cursor or from an operator's hand, because "the revision you
    /// asked to resume from is gone" and "there is nothing more" are different
    /// facts about a paginated history.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub next_start: Option<String>,
}

/// An actor, projected. The same three kinds the domain records, so an audit
/// reader can filter on `breakglass` without parsing prose.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ActorView {
    pub kind: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub issuer: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub subject: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub component: Option<String>,
}

impl ActorView {
    pub fn of(actor: &Actor) -> Self {
        match actor {
            Actor::Human { issuer, subject } => Self {
                kind: "human",
                issuer: Some(issuer.clone()),
                subject: Some(subject.clone()),
                component: None,
            },
            Actor::Breakglass => Self {
                kind: "breakglass",
                issuer: None,
                subject: None,
                component: None,
            },
            Actor::System { component } => Self {
                kind: "system",
                issuer: None,
                subject: None,
                component: Some(component.clone()),
            },
        }
    }
}

/// One audit event, projected.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AuditRecord {
    pub event: String,
    pub mutation: String,
    pub actor: ActorView,
    pub kind: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub target: Option<String>,
    /// The summary the mutation carried. Bounded and printable at submission by
    /// [`AuditSummary`](super::protocol::AuditSummary), and prefixed with the
    /// attribution when breakglass published it.
    pub summary: String,
    pub recorded_at_ms: u64,
}

impl AuditRecord {
    pub fn of(event: &AuditEvent) -> Self {
        Self {
            event: event.id.to_string(),
            mutation: event.mutation.to_string(),
            actor: ActorView::of(&event.actor),
            kind: event.kind.as_str(),
            target: event.target.map(|target| target.to_string()),
            summary: event.summary.clone(),
            recorded_at_ms: millis(event.recorded_at),
        }
    }
}

/// A revision's audit trail, bounded.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AuditPage {
    pub revision: String,
    pub events: Vec<AuditRecord>,
    /// Whether the store held more events than [`AuditPage::MAX_EVENTS`].
    pub truncated: bool,
}

impl AuditPage {
    pub const MAX_EVENTS: usize = 100;

    pub fn of(revision: RevisionId, events: &[AuditEvent]) -> Self {
        Self {
            revision: revision.to_string(),
            events: events
                .iter()
                .take(Self::MAX_EVENTS)
                .map(AuditRecord::of)
                .collect(),
            truncated: events.len() > Self::MAX_EVENTS,
        }
    }
}

/// One resource, as a state read shows it: everything except its body.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ResourceRecord {
    pub kind: &'static str,
    pub resource: String,
    pub version: u64,
    pub scope: ScopeView,
    pub slug: String,
    /// The checksum of the resource version's canonical bytes.
    pub content: String,
    /// The exact resource versions this one requires, in stable order.
    pub depends_on: Vec<String>,
}

/// A blob a revision declares. References only: the payload is not projected,
/// and there is no administrative read that returns one.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct BlobRecord {
    pub kind: &'static str,
    pub digest: String,
    pub size_bytes: u64,
}

/// The complete desired state, projected.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct StateView {
    /// `None` before the first publication: an empty control plane is not an
    /// error.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub revision: Option<String>,
    pub resources: Vec<ResourceRecord>,
    pub blobs: Vec<BlobRecord>,
}

impl StateView {
    pub fn of(revision: Option<&LoadedRevision>) -> Result<Self, AdminError> {
        let empty = DesiredState::new();
        let state = revision.map_or(&empty, LoadedRevision::state);
        let mut resources = Vec::with_capacity(state.resources().len());
        for resource in state.resources() {
            resources.push(ResourceRecord {
                kind: resource.reference.kind.as_str(),
                resource: resource.reference.id.to_string(),
                version: resource.reference.version.get(),
                scope: ScopeView::of(&resource.scope),
                slug: resource.slug.as_str().to_owned(),
                content: resource.content_checksum()?.to_string(),
                depends_on: resource
                    .depends_on
                    .iter()
                    .map(ToString::to_string)
                    .collect(),
            });
        }
        Ok(Self {
            revision: revision.map(|revision| revision.id().to_string()),
            resources,
            blobs: state
                .blobs()
                .map(|blob| BlobRecord {
                    kind: blob.kind.as_str(),
                    digest: blob.digest.to_string(),
                    size_bytes: blob.size_bytes,
                })
                .collect(),
        })
    }
}

/// What this replica has converged onto, and why it has not.
///
/// The contract later resource handlers report after a publication: an
/// administrator who published revision R wants to know whether R is being served
/// yet, and if not, whether that is normal lag or a refusal.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ConvergenceResult {
    pub converged: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub desired: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub loaded: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub active: Option<String>,
    /// `control-plane` or `last-known-good`: "serving R" means something
    /// different when R came from a cache at boot.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub source: Option<&'static str>,
    pub generation: u64,
    pub lag_ms: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_convergence_ms: Option<u64>,
    pub consecutive_failures: u32,
    /// The low-cardinality reason the last candidate was refused. The rejection's
    /// operator detail is deliberately not projected: it is a log field, and this
    /// is a response.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_rejection: Option<&'static str>,
}

impl ConvergenceResult {
    pub fn of(report: &RevisionReport) -> Self {
        Self {
            converged: report.converged(),
            desired: report.desired.map(|id| id.to_string()),
            loaded: report.loaded.map(|id| id.to_string()),
            active: report.active.map(|id| id.to_string()),
            source: report.source.map(|source| source.as_str()),
            generation: report.generation,
            lag_ms: duration_ms(report.lag),
            last_convergence_ms: report.last_convergence.map(duration_ms),
            consecutive_failures: report.consecutive_failures,
            last_rejection: report
                .last_rejection
                .as_ref()
                .map(|rejection| rejection.reason),
        }
    }
}

/// Milliseconds since the Unix epoch, saturating rather than failing: a wrong
/// host clock must not make an audit trail unreadable.
fn millis(at: SystemTime) -> u64 {
    at.duration_since(UNIX_EPOCH).map_or(0, |since| {
        u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
    })
}

fn duration_ms(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}