hyperhive/swarm-wire/src/router_api.rs

//! What the notification router records and answers, for swarm-controller's
//! `GET /api/notify/deliveries` and `GET /api/notify/dry-run`.
//!
//! The router appends one [`Row`] to the [`LOG_STREAM`] stream for each
//! delivery it makes, and keeps the newest [`LOG_MAX_MSGS`]. The controller
//! asks it two things as NATS requests: the newest rows ([`DELIVERIES_SUBJECT`])
//! and, for a flows document, how many of the newest forge events reach each
//! node ([`DRY_RUN_SUBJECT`]), run without delivering anything.
//!
//! Three crates name these: the router records and answers, the controller
//! asks, and the auth-callout responder grants each its side. A router that
//! predates this answers neither, so the controller reports that.

use std::time::Duration;

use serde::{Deserialize, Serialize};

use crate::notify_flows::Document;

/// The stream holding the deliveries log.
pub const LOG_STREAM: &str = "NOTIFY_ROUTER_LOG";

/// The one subject [`LOG_STREAM`] captures. The router is its one publisher.
pub const LOG_SUBJECT: &str = "$SWARM.router.log";

/// How many rows the stream keeps; the oldest goes first.
pub const LOG_MAX_MSGS: i64 = 500;

/// The request for the newest [`Row`]s ([`DeliveriesRequest`]).
pub const DELIVERIES_SUBJECT: &str = "$SWARM.router.api.deliveries";

/// The request to run a document over the newest forge events
/// ([`DryRunRequest`]).
pub const DRY_RUN_SUBJECT: &str = "$SWARM.router.api.dry-run";

/// How many of the newest forge events a dry run covers.
pub const DRY_RUN_EVENTS: usize = 50;

/// The rows a request gets when it names no limit.
pub const DEFAULT_ROWS: usize = 100;

/// How long the controller waits for the router. The router's reply grant
/// lasts this long.
pub const REPLY_DEADLINE: Duration = Duration::from_secs(10);

/// What a delivery was.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
#[serde(rename_all = "snake_case")]
pub enum Kind {
    /// A todo op queued for an agent.
    Todo,
    /// A request to the controller for a review.
    RequestReviewer,
}

/// One delivery the router made.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct Row {
    /// Unix seconds.
    pub at: u64,
    pub flow: String,
    pub node: String,
    /// The forge event type, such as `issue_assign`.
    pub event: String,
    /// `owner/name` of the repository the event concerns.
    pub repo: String,
    /// The issue or pull request number.
    pub number: u64,
    pub kind: Kind,
    /// The agent that got the todo, or the reviewer asked for.
    pub target: String,
    /// Whether it went through. A review the controller skipped went through.
    pub ok: bool,
    /// What happened: the op and key, the controller's answer, or the error.
    pub detail: String,
}

/// Ask for the newest rows, newest first.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct DeliveriesRequest {
    /// At most this many; [`DEFAULT_ROWS`] when absent. The log keeps
    /// [`LOG_MAX_MSGS`], so no answer is longer.
    #[serde(default)]
    pub limit: Option<usize>,
}

/// The router's answer to a [`DeliveriesRequest`].
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct Deliveries {
    /// Newest first.
    pub rows: Vec<Row>,
}

/// Run `document` over the newest [`DRY_RUN_EVENTS`] forge events.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DryRunRequest {
    pub document: Document,
}

/// How many events reached one node.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct NodeHits {
    pub node: String,
    /// Events that reached the node. A `match` node counts those that passed
    /// it, a `source` node those it starts from.
    pub hits: u64,
}

/// The hits of one flow that would run.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct FlowHits {
    pub flow: String,
    /// Every node of the flow, those no event reached with `0`.
    pub nodes: Vec<NodeHits>,
}

/// The router's answer to a [`DryRunRequest`] it ran.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct DryRun {
    /// How many events were run, at most [`DRY_RUN_EVENTS`]: fewer when the
    /// stream holds fewer.
    pub events: u64,
    /// The flows the document turns on: the built-ins it leaves on and its
    /// custom flows that are enabled.
    pub flows: Vec<FlowHits>,
}

/// The router's answer to a request it could not serve.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Refused {
    pub error: String,
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The log keeps the rows the deliveries view can ask for.
    #[test]
    fn a_request_never_asks_for_more_than_the_stream_keeps() {
        assert!(i64::try_from(DEFAULT_ROWS).expect("small") <= LOG_MAX_MSGS);
    }

    #[test]
    fn a_request_without_a_limit_decodes() {
        let r: DeliveriesRequest = serde_json::from_str("{}").expect("empty request");
        assert_eq!(r.limit, None);
    }

    #[test]
    fn the_reply_deadline_leaves_the_controller_time_to_answer() {
        assert!(REPLY_DEADLINE < Duration::from_secs(30));
    }
}