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