From bb08fac33cd1f8749cb8244f21bc2ab6884908ef Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 24 Oct 2026 03:44:00 +0200 Subject: [PATCH] notify router: read custom flows and disabled built-ins from a stored document --- swarm-controller/src/main.rs | 1 + swarm-controller/src/notify_flows.rs | 232 ++++++++ swarm-controller/src/startup.rs | 1 + swarm-nats-auth/src/policy.rs | 20 +- swarm-notify-router/src/event.rs | 16 + swarm-notify-router/src/flow.rs | 428 ++++----------- swarm-notify-router/src/main.rs | 25 +- swarm-notify-router/src/route.rs | 8 +- swarm-notify-router/src/stored.rs | 254 +++++++++ swarm-wire/src/lib.rs | 1 + swarm-wire/src/notify_flows.rs | 774 +++++++++++++++++++++++++++ 11 files changed, 1418 insertions(+), 342 deletions(-) create mode 100644 swarm-controller/src/notify_flows.rs create mode 100644 swarm-notify-router/src/stored.rs create mode 100644 swarm-wire/src/notify_flows.rs diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index a5434dd2..7439a410 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -50,6 +50,7 @@ mod jobq; mod linked_accounts; mod matrix_account; mod message_stream; +mod notify_flows; mod otel_http_client; mod pulls; mod queue_identity; diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs new file mode 100644 index 00000000..ad03c915 --- /dev/null +++ b/swarm-controller/src/notify_flows.rs @@ -0,0 +1,232 @@ +//! `GET` and `PUT /api/notify/flows`: which of the notification router's +//! built-in flows are switched off, and the custom flows. +//! +//! The flows are one document under one key ([`notify_flows::KEY`] in +//! [`notify_flows::BUCKET`]), which only this daemon writes, as the schedule +//! definitions are. A `PUT` replaces the whole document in one +//! compare-and-set, after [`Document::check`]. The router reads the key +//! every few seconds, so a change applies to the next event with no restart. +//! No document means the built-in flows alone. +//! +//! The built-in flows are defined in code, in `swarm_wire::notify_flows`, and +//! the document stores only their disabled ids, so a `GET` shows the ones +//! this build knows. + +use anyhow::Context as _; +use async_nats::jetstream::kv::Store; +use axum::Json; +use axum::extract::State; +use axum::extract::rejection::JsonRejection; +use axum::http::StatusCode; +use problem_details::ProblemDetails; +use serde::{Deserialize, Serialize}; +use swarm_wire::notify_flows::{self, DecodeError, Document, Flow}; +use utoipa::ToSchema; + +use crate::routes::{AppState, error_problem, wanted_error_status}; +use crate::schedules::{queue, read_with, refused, write_revision}; + +/// How many times a losing writer re-reads and re-applies, as in +/// `wanted::MAX_ATTEMPTS`. +const MAX_ATTEMPTS: usize = 5; + +/// What `GET` and a successful `PUT` answer. +#[derive(Debug, Clone, PartialEq, Serialize, ToSchema)] +pub(crate) struct FlowsView { + /// The revision of the stored document, to send back in a `PUT`. `null` + /// when none is stored. + revision: Option, + /// The ids of the built-in flows that are switched off. + disabled_builtins: Vec, + /// The built-in flows this build ships, all of them, switched off or not. + builtins: Vec, + /// The custom flows. + flows: Vec, +} + +/// The body of a `PUT`: the whole document. +#[derive(Debug, Clone, Deserialize, ToSchema)] +#[serde(deny_unknown_fields)] +pub(crate) struct PutFlows { + /// The `revision` the client read. When another write has landed since, + /// the `PUT` is refused. Left out, it replaces whatever is stored. + #[serde(default)] + revision: Option, + /// The ids of the built-in flows to switch off. + #[serde(default)] + disabled_builtins: Vec, + /// The custom flows. + #[serde(default)] + flows: Vec, +} + +impl FlowsView { + fn of(doc: Document, revision: Option) -> Self { + Self { + revision, + disabled_builtins: doc.disabled_builtins, + builtins: notify_flows::builtins(), + flows: doc.flows, + } + } +} + +/// What the key holds. +enum Stored { + /// No key, or a deleted one. + Nothing, + Document(Document), + /// A document this build cannot read, with why. + Unreadable(DecodeError), +} + +/// The stored document, with the revision to write against. +async fn stored(store: &Store) -> anyhow::Result<(Stored, Option)> { + struct Read(Stored); + impl Default for Read { + fn default() -> Self { + Self(Stored::Nothing) + } + } + let (Read(stored), revision) = read_with(store, notify_flows::KEY, |bytes| { + Ok(Read(match Document::decode(bytes) { + Ok(doc) => Stored::Document(doc), + Err(e) => Stored::Unreadable(e), + })) + }) + .await?; + Ok((stored, revision)) +} + +/// Open [`notify_flows::BUCKET`], creating it if nothing has yet. +async fn open(client: &async_nats::Client) -> anyhow::Result { + let js = async_nats::jetstream::new(client.clone()); + if let Ok(store) = js.get_key_value(notify_flows::BUCKET).await { + return Ok(store); + } + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: notify_flows::BUCKET.to_owned(), + description: "The notification router's flows: disabled built-ins and custom flows" + .to_owned(), + history: 1, + ..Default::default() + }) + .await + .with_context(|| format!("creating the {} bucket", notify_flows::BUCKET)) +} + +fn queue_problem(e: &anyhow::Error) -> ProblemDetails { + tracing::warn!(error = %format!("{e:#}"), "notify flows: queue failure"); + error_problem(wanted_error_status(e), &format!("{e:#}")) +} + +/// Read the flows. +#[utoipa::path( + get, + path = "/api/notify/flows", + responses( + (status = 200, description = "the stored document, with the built-in flows this \ + build ships; with none stored, no disabled built-ins and no custom flows", + body = FlowsView), + (status = 409, description = "the stored document is not one this build reads, \ + such as a newer build's version (problem+json)", body = String), + (status = 503, description = "no swarm queue is configured on this host, or it is \ + not connected (problem+json)", body = String), + (status = 500, description = "the flows could not be read (problem+json)", + body = String), + ), + tag = "notify" +)] +pub(crate) async fn get_flows( + State(state): State, +) -> Result, ProblemDetails> { + let client = queue(&state).map_err(refused)?; + let store = open(&client).await.map_err(|e| queue_problem(&e))?; + let (stored, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; + match stored { + Stored::Nothing => Ok(Json(FlowsView::of(Document::default(), revision))), + Stored::Document(doc) => Ok(Json(FlowsView::of(doc, revision))), + Stored::Unreadable(e) => Err(unreadable(&e)), + } +} + +fn unreadable(why: &DecodeError) -> ProblemDetails { + error_problem( + StatusCode::CONFLICT, + &format!("the stored flows document is not one this build reads: {why}"), + ) +} + +/// Replace the flows. +#[utoipa::path( + put, + path = "/api/notify/flows", + request_body = PutFlows, + responses( + (status = 200, description = "the document as stored; the router applies it from \ + its next event", body = FlowsView), + (status = 400, description = "not JSON or not this shape, or the document fails its \ + check: a `disabled_builtins` entry that is no built-in, a custom flow whose id \ + is another's or a built-in's, a flow that is not a tree (one `source`, one \ + parent for every other node, every node reachable, no children under a \ + `todo`), more than 64 nodes, 16 levels or 100 flows, an id that is not 1 to 64 \ + letters, digits, `-` or `_`, a source event the router is not sent, a condition \ + or recipient field that is no event field, or an upsert with no text \ + (problem+json)", body = String), + (status = 409, description = "another write landed since `revision`, or the stored \ + document is a version this build does not read, which a `PUT` would erase \ + (problem+json)", body = String), + (status = 503, description = "no swarm queue is configured on this host, or it is \ + not connected (problem+json)", body = String), + (status = 500, description = "the flows could not be written (problem+json)", + body = String), + ), + tag = "notify" +)] +pub(crate) async fn put_flows( + State(state): State, + body: Result, JsonRejection>, +) -> Result, ProblemDetails> { + let Json(put) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; + let doc = Document { + version: notify_flows::VERSION, + disabled_builtins: put.disabled_builtins, + flows: put.flows, + }; + doc.check() + .map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?; + let client = queue(&state).map_err(refused)?; + let store = open(&client).await.map_err(|e| queue_problem(&e))?; + for _ in 0..MAX_ATTEMPTS { + let (current, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; + // An unreadable document is replaced only when it is not a version + // this build does not know: a newer build's document is kept for it. + if let Stored::Unreadable(why @ DecodeError::UnknownVersion(_)) = ¤t + { + return Err(unreadable(why)); + } + if put.revision.is_some() && put.revision != revision { + return Err(error_problem( + StatusCode::CONFLICT, + &format!( + "the flows changed since revision {}; read them again", + put.revision.unwrap_or_default() + ), + )); + } + if let Some(written) = write_revision(&store, notify_flows::KEY, &doc, revision) + .await + .map_err(|e| queue_problem(&e))? + { + tracing::info!( + disabled = ?doc.disabled_builtins, + custom = doc.flows.len(), + "set the notification router's flows" + ); + return Ok(Json(FlowsView::of(doc, Some(written)))); + } + } + Err(queue_problem(&anyhow::anyhow!( + "gave up writing the notification flows after {MAX_ATTEMPTS} conflicting writes" + ))) +} diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index 29a0b418..b6eca2a1 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -635,6 +635,7 @@ pub(crate) fn build_app(state: AppState) -> axum::Router { review_classifier::settings::get_settings, review_classifier::settings::put_settings )) + .routes(routes!(notify_flows::get_flows, notify_flows::put_flows)) .routes(routes!(webhook::post_webhook_forge)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index f294dde1..000dbf97 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -317,7 +317,7 @@ impl Policy { /// What the notification router may do: drain its own consumer on the /// forge events stream, queue todo ops for any agent, keep its own state - /// bucket, and list which agents exist. + /// bucket, read the flows document, and list which agents exist. /// /// - **Forge events.** Open or create the stream, create its one durable /// in the extended form whose last tokens are the filter (the server @@ -330,14 +330,18 @@ impl Policy { /// subjects, which is what makes the router their sole writer. /// - **Its state bucket.** Create, look up, read and write /// `notify-router-state`, and no other bucket. + /// - **The flows document.** `STREAM.INFO` and `DIRECT.GET` on + /// `notify-flows`, which it reads and never creates or writes: the + /// controller creates the bucket and is its one writer. /// - **The roster.** `STREAM.INFO` on the agent-status bucket, which lists /// its keys (`.`) and writes none. fn notify_router_permissions(client_id: &str) -> Permissions { - use swarm_wire::{forge_events, notify_router, router_todo}; + use swarm_wire::{forge_events, notify_flows, notify_router, router_todo}; let events = forge_events::STREAM; let consumer = forge_events::ROUTER_CONSUMER; let state = format!("KV_{}", notify_router::STATE_BUCKET); + let flows = format!("KV_{}", notify_flows::BUCKET); let mut publish = Self::jetstream_minimum().to_vec(); publish.extend([ format!("$JS.API.STREAM.INFO.{events}"), @@ -355,6 +359,8 @@ impl Policy { format!("$JS.API.STREAM.CREATE.{state}"), format!("$JS.API.DIRECT.GET.{state}.>"), format!("$KV.{}.>", notify_router::STATE_BUCKET), + format!("$JS.API.STREAM.INFO.{flows}"), + format!("$JS.API.DIRECT.GET.{flows}.>"), format!("$JS.API.STREAM.INFO.{}", Self::agent_status_stream()), Self::inbox(client_id), ]); @@ -1799,7 +1805,7 @@ mod tests { #[test] fn the_notify_router_drains_its_own_consumer_and_writes_only_its_own_state() { use swarm_queue_client::client_inbox_prefix; - use swarm_wire::{forge_events, notify_router, router_todo}; + use swarm_wire::{forge_events, notify_flows, notify_router, router_todo}; let p = with_router() .permissions(NOTIFY_ROUTER) @@ -1807,6 +1813,7 @@ mod tests { let may = |subject: &str| p.publish.iter().any(|g| covers(g, subject)); let events = forge_events::STREAM; let state = notify_router::STATE_BUCKET; + let flows = notify_flows::BUCKET; for allowed in [ format!( "$JS.API.CONSUMER.CREATE.{events}.notify-router.{}", @@ -1820,6 +1827,9 @@ mod tests { format!("$KV.{state}.42.5571"), format!("$JS.API.DIRECT.GET.KV_{state}.$KV.{state}.42.5571"), "$JS.API.STREAM.INFO.KV_agent-status".to_owned(), + // The flows document: read it. + format!("$JS.API.STREAM.INFO.KV_{flows}"), + format!("$JS.API.DIRECT.GET.KV_{flows}.$KV.{flows}.{}", notify_flows::KEY), ] { assert!(may(&allowed), "the router may not reach {allowed}"); } @@ -1838,6 +1848,10 @@ mod tests { forge_events::subject("issues"), "$KV.agent-status.alpha.atlas".to_owned(), "$KV.agent-schedules.atlas".to_owned(), + // The flows document: not write it, nor make the bucket. + format!("$KV.{flows}.{}", notify_flows::KEY), + format!("$JS.API.STREAM.CREATE.KV_{flows}"), + format!("$JS.API.STREAM.DELETE.KV_{flows}"), swarm_queue_client::agent_input_subject("atlas"), "$JS.API.STREAM.CREATE.KV_agent-status".to_owned(), ] { diff --git a/swarm-notify-router/src/event.rs b/swarm-notify-router/src/event.rs index c7a80317..b2ef0f46 100644 --- a/swarm-notify-router/src/event.rs +++ b/swarm-notify-router/src/event.rs @@ -163,6 +163,22 @@ mod tests { } } + /// The names a stored flow may read are the ones `fields` produces. + #[test] + fn the_fields_are_the_ones_a_stored_flow_may_name() { + let d = Delivery::parse(&issue_assign("assigned", &["atlas"])).expect("parses"); + for event in ["issues", "issue_assign"] { + let mut got: Vec<&str> = fields(event, &d, &Snapshot::default()) + .keys() + .copied() + .collect(); + let mut want = swarm_wire::notify_flows::FIELDS.to_vec(); + got.sort_unstable(); + want.sort_unstable(); + assert_eq!(got, want, "{event}"); + } + } + #[test] fn an_issue_delivery_reads_back_its_fields() { let d = Delivery::parse(&issue_assign("assigned", &["atlas", "iris"])).expect("parses"); diff --git a/swarm-notify-router/src/flow.rs b/swarm-notify-router/src/flow.rs index 89a2ca2a..3eca668d 100644 --- a/swarm-notify-router/src/flow.rs +++ b/swarm-notify-router/src/flow.rs @@ -1,26 +1,17 @@ -//! Flows: what the router does with an event. +//! Running a flow over one event. //! -//! A flow is a graph of nodes joined by edges, rooted at one `source` node. -//! A `match` node passes when every one of its conditions holds, and then -//! each of its children runs, so one event can reach several branches. A -//! `sink` node is a leaf and emits one todo op per recipient. -//! -//! Flows are trees: one source, one parent for every other node, every node -//! reachable from the source. [`Flow::validate`] checks that, and the router -//! runs only flows that pass. -//! -//! The built-in flows ([`builtins`]) are ordinary flows defined in code. +//! The flow types, their checks and the built-in flows are +//! `swarm_wire::notify_flows`; the built-in and the custom flows are the same +//! type and this walk runs both. A `match` node passes when every one of its +//! conditions holds, and then each of its children runs, so one event can +//! reach several branches. A `todo` node emits one op per recipient. The +//! router runs only flows that pass [`Flow::validate`]. -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::BTreeMap; +use swarm_wire::notify_flows::{Condition, Flow, Node, NodeKind, Recipients, TodoSink}; use swarm_wire::router_todo::{Op, RouterTodoOp}; -/// The most nodes one flow may have. -const MAX_NODES: usize = 64; - -/// The deepest a node may sit below the source. -const MAX_DEPTH: usize = 16; - /// One event, as the fields a flow reads. pub type Fields = BTreeMap<&'static str, Value>; @@ -31,56 +22,6 @@ pub enum Value { List(Vec), } -#[derive(Clone, Debug)] -pub struct Flow { - pub id: String, - pub nodes: Vec, - pub edges: Vec, -} - -#[derive(Clone, Debug)] -pub struct Node { - pub id: String, - pub kind: NodeKind, -} - -/// A directed edge, parent to child. -#[derive(Clone, Debug)] -pub struct Edge { - pub from: String, - pub to: String, -} - -#[derive(Clone, Debug)] -pub enum NodeKind { - /// The root: the forge event types the flow starts from. - Source { events: Vec }, - /// Passes when every condition holds. - Match { when: Vec }, - /// Emits a todo op to each recipient. - Todo(TodoSink), -} - -/// `field` equals `value`. A list field holds when one of its items does. -#[derive(Clone, Debug)] -pub struct Condition { - pub field: String, - pub equals: String, -} - -#[derive(Clone, Debug)] -pub struct TodoSink { - /// The field naming the recipients, a list of agent names. - pub to: String, - pub op: Op, - /// The todo key, with `{field}` placeholders. - pub key: String, - /// For a clear: `key` is a prefix. - pub prefix: bool, - /// The todo text, with `{field}` placeholders. Unused for a clear. - pub text: String, -} - /// One op the walk emitted, for one recipient. #[derive(Clone, Debug, PartialEq, Eq)] pub struct Action { @@ -90,130 +31,61 @@ pub struct Action { pub op: RouterTodoOp, } -impl Flow { - /// Check the flow is a tree within the caps, or say why not. - pub fn validate(&self) -> Result<(), String> { - if self.nodes.len() > MAX_NODES { - return Err(format!("{} nodes, more than {MAX_NODES}", self.nodes.len())); - } - let mut ids = BTreeSet::new(); - for node in &self.nodes { - if !ids.insert(node.id.as_str()) { - return Err(format!("node id {} is used twice", node.id)); - } - } - for edge in &self.edges { - for end in [&edge.from, &edge.to] { - if !ids.contains(end.as_str()) { - return Err(format!("an edge names {end}, which is no node")); +/// Walk `flow` for one event of type `event_type` and return the ops it +/// emits. Assumes [`Flow::validate`] passed. +pub fn evaluate(flow: &Flow, event_type: &str, fields: &Fields) -> Vec { + let Some(root) = flow + .nodes + .iter() + .find(|n| matches!(&n.kind, NodeKind::Source { events } if events.iter().any(|e| e == event_type))) + else { + return Vec::new(); + }; + let mut actions = Vec::new(); + let mut stack = vec![root.id.as_str()]; + while let Some(id) = stack.pop() { + let Some(node) = flow.node(id) else { continue }; + match &node.kind { + NodeKind::Source { .. } => stack.extend(flow.children(id)), + NodeKind::Match { when } => { + if when.iter().all(|c| holds(c, fields)) { + stack.extend(flow.children(id)); } } + NodeKind::Todo(sink) => actions.extend(emit(flow, node, sink, fields)), } - let roots: Vec<&Node> = self - .nodes - .iter() - .filter(|n| matches!(n.kind, NodeKind::Source { .. })) - .collect(); - let [root] = roots.as_slice() else { - return Err(format!("{} source nodes, not one", roots.len())); - }; - for node in &self.nodes { - if matches!(node.kind, NodeKind::Todo(_)) && !self.children(&node.id).is_empty() { - return Err(format!("sink {} has children", node.id)); - } - } - // From the one root, every node reached exactly once: a node reached - // twice has a second parent or sits on a cycle, and one not reached is - // cut off. - let mut reached = 0; - let mut stack = vec![(root.id.as_str(), 0)]; - while let Some((id, depth)) = stack.pop() { - if depth > MAX_DEPTH { - return Err(format!("node {id} is deeper than {MAX_DEPTH}")); - } - reached += 1; - if reached > self.nodes.len() { - return Err( - "a node is reached twice: it has two parents or sits on a cycle".to_owned(), - ); - } - stack.extend(self.children(id).into_iter().map(|c| (c, depth + 1))); - } - if reached != self.nodes.len() { - return Err(format!( - "{} of {} nodes are not reachable from the source", - self.nodes.len() - reached, - self.nodes.len() - )); - } - Ok(()) } + actions +} - fn children(&self, id: &str) -> Vec<&str> { - self.edges - .iter() - .filter(|e| e.from == id) - .map(|e| e.to.as_str()) - .collect() - } - - fn node(&self, id: &str) -> Option<&Node> { - self.nodes.iter().find(|n| n.id == id) - } - - /// Walk the flow for one event of type `event_type` and return the ops it - /// emits. Assumes [`Self::validate`] passed. - pub fn evaluate(&self, event_type: &str, fields: &Fields) -> Vec { - let Some(root) = self - .nodes - .iter() - .find(|n| matches!(&n.kind, NodeKind::Source { events } if events.iter().any(|e| e == event_type))) - else { - return Vec::new(); - }; - let mut actions = Vec::new(); - let mut stack = vec![root.id.as_str()]; - while let Some(id) = stack.pop() { - let Some(node) = self.node(id) else { continue }; - match &node.kind { - NodeKind::Source { .. } => stack.extend(self.children(id)), - NodeKind::Match { when } => { - if when.iter().all(|c| holds(c, fields)) { - stack.extend(self.children(id)); - } - } - NodeKind::Todo(sink) => actions.extend(self.emit(node, sink, fields)), - } - } - actions - } - - fn emit(&self, node: &Node, sink: &TodoSink, fields: &Fields) -> Vec { - let recipients = match fields.get(sink.to.as_str()) { +fn emit(flow: &Flow, node: &Node, sink: &TodoSink, fields: &Fields) -> Vec { + let recipients = match &sink.to { + Recipients::Agent(agent) => vec![agent.clone()], + Recipients::Field(name) => match fields.get(name.as_str()) { Some(Value::List(items)) => items.clone(), Some(Value::Text(one)) => vec![one.clone()], None => Vec::new(), - }; - let op = RouterTodoOp { - op: sink.op, - key: expand(&sink.key, fields), - prefix: sink.op == Op::Clear && sink.prefix, - summary: match sink.op { - Op::Upsert => expand(&sink.text, fields), - Op::Clear => String::new(), - }, - source: format!("router:{}", self.id), - }; - recipients - .into_iter() - .map(|agent| Action { - flow: self.id.clone(), - node: node.id.clone(), - agent, - op: op.clone(), - }) - .collect() - } + }, + }; + let op = RouterTodoOp { + op: sink.op, + key: expand(&sink.key, fields), + prefix: sink.op == Op::Clear && sink.prefix, + summary: match sink.op { + Op::Upsert => expand(&sink.text, fields), + Op::Clear => String::new(), + }, + source: format!("router:{}", flow.id), + }; + recipients + .into_iter() + .map(|agent| Action { + flow: flow.id.clone(), + node: node.id.clone(), + agent, + op: op.clone(), + }) + .collect() } fn holds(condition: &Condition, fields: &Fields) -> bool { @@ -252,91 +124,11 @@ pub fn expand(template: &str, fields: &Fields) -> String { out } -/// The flows the router ships with. -/// -/// - `assigned`: a newly assigned agent gets a todo for the issue or PR. -/// - `unassigned`: an agent taken off it gets that todo rewritten to say so. -/// Both write the same key, so the second replaces the first. -/// - `closed`: closing it clears every router todo the router queued for it. -pub fn builtins() -> Vec { - let assignment_key = "forge:{repo}#{number}:assignment"; - vec![ - chain( - "assigned", - &["issue_assign", "pull_request_assign"], - "assigned", - TodoSink { - to: "assignees_added".to_owned(), - op: Op::Upsert, - key: assignment_key.to_owned(), - prefix: false, - text: "you are assigned to {kind} {repo}#{number}: {title} ({url})".to_owned(), - }, - ), - chain( - "unassigned", - &["issue_assign", "pull_request_assign"], - "unassigned", - TodoSink { - to: "assignees_removed".to_owned(), - op: Op::Upsert, - key: assignment_key.to_owned(), - prefix: false, - text: "you are not assigned to {kind} {repo}#{number} anymore: {title} ({url})" - .to_owned(), - }, - ), - chain( - "closed", - &["issues", "pull_request"], - "closed", - TodoSink { - to: "todo_holders".to_owned(), - op: Op::Clear, - key: "forge:{repo}#{number}:".to_owned(), - prefix: true, - text: String::new(), - }, - ), - ] -} - -/// source(`events`) → match(`action` = `action`) → `sink`. -fn chain(id: &str, events: &[&str], action: &str, sink: TodoSink) -> Flow { - let node = |id: &str, kind| Node { - id: id.to_owned(), - kind, - }; - let edge = |from: &str, to: &str| Edge { - from: from.to_owned(), - to: to.to_owned(), - }; - Flow { - id: id.to_owned(), - nodes: vec![ - node( - "source", - NodeKind::Source { - events: events.iter().map(|e| (*e).to_owned()).collect(), - }, - ), - node( - "match", - NodeKind::Match { - when: vec![Condition { - field: "action".to_owned(), - equals: action.to_owned(), - }], - }, - ), - node("sink", NodeKind::Todo(sink)), - ], - edges: vec![edge("source", "match"), edge("match", "sink")], - } -} #[cfg(test)] mod tests { + use swarm_wire::notify_flows::{Edge, builtins}; + use super::*; fn text(v: &str) -> Value { @@ -368,16 +160,11 @@ mod tests { .expect("a built-in") } - #[test] - fn every_builtin_is_a_valid_tree() { - for f in builtins() { - assert_eq!(f.validate(), Ok(()), "{}", f.id); - } - } - #[test] fn assigning_queues_a_todo_for_each_newly_assigned_agent_only() { - let got = flow("assigned").evaluate( + let assigned = flow("assigned"); + let got = evaluate( + &assigned, "issue_assign", &event("assigned", &["atlas", "iris"], &[], &[]), ); @@ -392,21 +179,28 @@ mod tests { assert_eq!(got[0].op.source, "router:assigned"); // The match gates: an unassign event reaches no assigned sink. assert!( - flow("assigned") - .evaluate("issue_assign", &event("unassigned", &["atlas"], &[], &[])) - .is_empty() + evaluate( + &assigned, + "issue_assign", + &event("unassigned", &["atlas"], &[], &[]) + ) + .is_empty() ); // The source gates: another event type starts nothing. assert!( - flow("assigned") - .evaluate("issues", &event("assigned", &["atlas"], &[], &[])) - .is_empty() + evaluate( + &assigned, + "issues", + &event("assigned", &["atlas"], &[], &[]) + ) + .is_empty() ); } #[test] fn unassigning_rewrites_the_same_key_for_the_removed_agent() { - let got = flow("unassigned").evaluate( + let got = evaluate( + &flow("unassigned"), "pull_request_assign", &event("unassigned", &[], &["argus"], &[]), ); @@ -423,8 +217,12 @@ mod tests { #[test] fn closing_clears_by_prefix_for_every_holder() { - let got = - flow("closed").evaluate("issues", &event("closed", &[], &[], &["atlas", "argus"])); + let closed = flow("closed"); + let got = evaluate( + &closed, + "issues", + &event("closed", &[], &[], &["atlas", "argus"]), + ); assert_eq!(got.len(), 2); for a in &got { assert_eq!(a.op.op, Op::Clear); @@ -433,9 +231,12 @@ mod tests { assert_eq!(a.op.summary, ""); } assert!( - flow("closed") - .evaluate("issues", &event("reopened", &[], &[], &["atlas"])) - .is_empty() + evaluate( + &closed, + "issues", + &event("reopened", &[], &[], &["atlas"]) + ) + .is_empty() ); } @@ -450,7 +251,7 @@ mod tests { f.nodes.push(Node { id: "second".to_owned(), kind: NodeKind::Todo(TodoSink { - to: "assignees_removed".to_owned(), + to: Recipients::Field("assignees_removed".to_owned()), ..sink }), }); @@ -459,7 +260,8 @@ mod tests { to: "second".to_owned(), }); assert_eq!(f.validate(), Ok(())); - let got = f.evaluate( + let got = evaluate( + &f, "issue_assign", &event("assigned", &["atlas"], &["argus"], &[]), ); @@ -471,40 +273,26 @@ mod tests { assert_eq!(nodes, [("second", "argus"), ("sink", "atlas")]); } + /// A custom flow's sink can name one agent, whatever the event holds. #[test] - fn a_flow_that_is_not_a_tree_is_refused() { - let edge = |from: &str, to: &str| Edge { - from: from.to_owned(), - to: to.to_owned(), - }; - let mut cycle = flow("assigned"); - cycle.edges[0] = edge("sink", "match"); - cycle.edges.push(edge("match", "sink")); - assert!(cycle.validate().is_err(), "a cycle"); - - let mut orphan = flow("assigned"); - orphan.edges.pop(); - assert!(orphan.validate().is_err(), "an unreachable sink"); - - let mut two_parents = flow("assigned"); - two_parents.edges.push(edge("source", "sink")); - assert!(two_parents.validate().is_err(), "a sink with two parents"); - - let mut sink_child = flow("assigned"); - sink_child.nodes.push(Node { - id: "m2".to_owned(), - kind: NodeKind::Match { when: Vec::new() }, + fn a_sink_to_a_named_agent_reaches_it_once() { + let mut f = flow("assigned"); + f.nodes[2].kind = NodeKind::Todo(TodoSink { + to: Recipients::Agent("triage".to_owned()), + op: Op::Upsert, + key: "forge:{repo}#{number}:triage".to_owned(), + prefix: false, + text: "triage {repo}#{number}".to_owned(), }); - sink_child.edges.push(edge("sink", "m2")); - assert!(sink_child.validate().is_err(), "a sink with a child"); - - let mut dangling = flow("assigned"); - dangling.edges.push(edge("match", "nowhere")); - assert!(dangling.validate().is_err(), "an edge to no node"); - - let mut two_sources = flow("assigned"); - two_sources.nodes[1].kind = NodeKind::Source { events: Vec::new() }; - assert!(two_sources.validate().is_err(), "two sources"); + let got = evaluate( + &f, + "issue_assign", + &event("assigned", &["atlas", "iris"], &[], &[]), + ); + assert_eq!(got.len(), 1); + assert_eq!(got[0].agent, "triage"); + assert_eq!(got[0].op.summary, "triage hyperhive/hyperhive#7"); + assert_eq!(got[0].op.source, "router:assigned"); } #[test] diff --git a/swarm-notify-router/src/main.rs b/swarm-notify-router/src/main.rs index 56d3b150..b108b0d9 100644 --- a/swarm-notify-router/src/main.rs +++ b/swarm-notify-router/src/main.rs @@ -4,7 +4,8 @@ //! (`swarm_wire::forge_events`), runs every flow over each delivery //! ([`flow`], [`route`]), and queues the todo ops the flows emit on each //! recipient agent's subject of `ROUTER_TODOS` (`swarm_wire::router_todo`). -//! Each agent's harness applies them to its own todo list. +//! Each agent's harness applies them to its own todo list. The flows are the +//! built-in ones, changed by the `notify-flows` document ([`stored`]). //! //! Per issue it keeps a snapshot in its state bucket ([`event::Snapshot`]): //! the assignees it last saw and the agents it queued a todo for. An event is @@ -31,6 +32,7 @@ mod event; mod flow; mod route; mod settle; +mod stored; /// The cert-auth role this service logs in to the store as. `swarm-bao.nix` /// writes it, matching the leaf's subject. @@ -83,17 +85,7 @@ async fn main() -> Result<()> { }) .await?; - let flows: Vec = flow::builtins() - .into_iter() - .filter(|f| match f.validate() { - Ok(()) => true, - Err(reason) => { - tracing::error!(flow = %f.id, %reason, "notify router: skipping a flow that is not a tree"); - false - } - }) - .collect(); - let mut router = Router::open(&queue, flows).await?; + let mut router = Router::open(&queue).await?; tracing::info!( stream = forge_events::STREAM, "notify router: routing forge events" @@ -113,12 +105,12 @@ struct Router { js: jetstream::Context, events: jetstream::consumer::Consumer, state: kv::Store, - flows: Vec, + flows: stored::Flows, roster: Option<(Instant, BTreeSet)>, } impl Router { - async fn open(queue: &async_nats::Client, flows: Vec) -> Result { + async fn open(queue: &async_nats::Client) -> Result { let js = jetstream::new(queue.clone()); let events = swarm_queue_client::forge_events::open_or_create(queue) .await @@ -154,7 +146,7 @@ impl Router { js, events, state, - flows, + flows: stored::Flows::builtins(), roster: None, }) } @@ -193,8 +185,9 @@ impl Router { }), None => event::Snapshot::default(), }; + self.flows.refresh(&self.js).await; let roster = self.roster().await.map_err(Skip::Retry)?; - let routed = route::route(&event_type, &delivery, &snapshot, &self.flows, |a| { + let routed = route::route(&event_type, &delivery, &snapshot, &self.flows.active, |a| { roster.contains(a) }); for action in &routed.actions { diff --git a/swarm-notify-router/src/route.rs b/swarm-notify-router/src/route.rs index a91a1705..a3395217 100644 --- a/swarm-notify-router/src/route.rs +++ b/swarm-notify-router/src/route.rs @@ -6,7 +6,9 @@ use std::collections::BTreeSet; use swarm_wire::router_todo::Op; use crate::event::{Delivery, Snapshot, fields, is_assign}; -use crate::flow::{Action, Flow}; +use swarm_wire::notify_flows::Flow; + +use crate::flow::{Action, evaluate}; /// What one event comes to. #[derive(Debug, PartialEq, Eq)] @@ -30,7 +32,7 @@ pub fn route( let fields = fields(event_type, delivery, snapshot); let mut actions = Vec::new(); let mut skipped = Vec::new(); - for action in flows.iter().flat_map(|f| f.evaluate(event_type, &fields)) { + for action in flows.iter().flat_map(|f| evaluate(f, event_type, &fields)) { if is_agent(&action.agent) { actions.push(action); } else if !skipped.contains(&action.agent) { @@ -68,7 +70,7 @@ pub fn route( #[cfg(test)] mod tests { use super::*; - use crate::flow::builtins; + use swarm_wire::notify_flows::builtins; fn delivery(action: &str, assignees: &[&str]) -> Delivery { Delivery { diff --git a/swarm-notify-router/src/stored.rs b/swarm-notify-router/src/stored.rs new file mode 100644 index 00000000..e34f834a --- /dev/null +++ b/swarm-notify-router/src/stored.rs @@ -0,0 +1,254 @@ +//! The flows the router runs: the built-ins, changed by the document in the +//! `notify-flows` bucket. +//! +//! The router reads the key again every [`REFRESH`] and runs what +//! [`Document::active`] says. What it does with each thing it can read: +//! +//! | the key | the router runs | +//! | ------------------------------------ | ----------------------------------- | +//! | absent, deleted, or no such bucket | the built-ins | +//! | a document it reads | its built-ins and custom flows | +//! | a document it cannot read | what it ran before | +//! | the read itself failing | what it ran before | +//! +//! "Cannot read" is not JSON, not the document's shape, or a version this +//! build does not know. Before any document has been read, "what it ran +//! before" is the built-ins, so a bad document never leaves the router with +//! no flows. + +use std::time::{Duration, Instant}; + +use async_nats::jetstream::context::GetStreamErrorKind; +use async_nats::jetstream::{self, kv}; +use async_nats::jetstream::ErrorCode; +use swarm_wire::notify_flows::{self, Document, Flow}; + +/// How long a read of the key is reused. +const REFRESH: Duration = Duration::from_secs(10); + +/// What a read of the key came to. +#[derive(Debug)] +pub enum Read { + /// No key, a deleted key, or no bucket. + Absent, + Document { revision: u64, bytes: Vec }, + Failed(String), +} + +/// The flows in force, and which revision of the key they came from. +pub struct Flows { + pub active: Vec, + /// The revision `active` was read from; `None` for the built-ins alone. + revision: Option, + /// The last problem logged, so a standing one is logged once. + problem: Option, + checked: Option, +} + +impl Flows { + /// The built-ins, which is what the router runs until it reads a document. + pub fn builtins() -> Self { + Self { + active: Document::default().active().0, + revision: None, + problem: None, + checked: None, + } + } + + /// Take what a read came to. + pub fn apply(&mut self, read: Read) { + match read { + Read::Failed(why) => self.note(format!("reading the key failed: {why}")), + Read::Absent => { + if self.revision.take().is_some() { + tracing::info!("notify router: the flows document is gone; running the built-in flows"); + } + self.active = Document::default().active().0; + self.problem = None; + } + Read::Document { revision, .. } if self.revision == Some(revision) => {} + Read::Document { revision, bytes } => match Document::decode(&bytes) { + Err(why) => { + // The revision is not taken, so the next read looks at + // the document again and the problem is logged once. + self.note(format!("revision {revision} cannot be read: {why}")); + } + Ok(doc) => { + let (active, skipped) = doc.active(); + for why in &skipped { + tracing::error!(%why, "notify router: skipping a flow"); + } + tracing::info!( + revision, + flows = active.len(), + disabled = doc.disabled_builtins.len(), + custom = doc.flows.len(), + "notify router: running the flows document" + ); + self.active = active; + self.revision = Some(revision); + self.problem = None; + } + }, + } + } + + fn note(&mut self, problem: String) { + if self.problem.as_ref() != Some(&problem) { + tracing::error!(%problem, "notify router: keeping the flows it has"); + self.problem = Some(problem); + } + } + + /// Read the key if the last read is older than [`REFRESH`]. + pub async fn refresh(&mut self, js: &jetstream::Context) { + if self.checked.is_some_and(|at| at.elapsed() < REFRESH) { + return; + } + self.checked = Some(Instant::now()); + self.apply(read(js).await); + } +} + +async fn read(js: &jetstream::Context) -> Read { + // A bucket that does not exist is the answer "no document". Any other + // failure to look, such as a timeout, says nothing about the document. + match js.get_stream(format!("KV_{}", notify_flows::BUCKET)).await { + Ok(_) => {} + Err(e) + if matches!(e.kind(), GetStreamErrorKind::JetStream(j) + if j.error_code() == ErrorCode::STREAM_NOT_FOUND) => + { + return Read::Absent; + } + Err(e) => return Read::Failed(e.to_string()), + } + let store = match js.get_key_value(notify_flows::BUCKET).await { + Ok(store) => store, + Err(e) => return Read::Failed(e.to_string()), + }; + match store.entry(notify_flows::KEY).await { + Ok(Some(entry)) if entry.operation == kv::Operation::Put => Read::Document { + revision: entry.revision, + bytes: entry.value.to_vec(), + }, + Ok(_) => Read::Absent, + Err(e) => Read::Failed(e.to_string()), + } +} + +#[cfg(test)] +mod tests { + use swarm_wire::notify_flows::{Document, VERSION}; + + use super::*; + + fn ids(f: &Flows) -> Vec<&str> { + f.active.iter().map(|f| f.id.as_str()).collect() + } + + fn doc(disabled: &[&str], custom: &[&str]) -> Read { + let mut flows = Vec::new(); + for id in custom { + let mut f = notify_flows::builtins().remove(0); + f.id = (*id).to_owned(); + flows.push(f); + } + let d = Document { + version: VERSION, + disabled_builtins: disabled.iter().map(|s| (*s).to_owned()).collect(), + flows, + }; + Read::Document { + revision: 0, + bytes: serde_json::to_vec(&d).unwrap(), + } + } + + fn at(revision: u64, read: Read) -> Read { + match read { + Read::Document { bytes, .. } => Read::Document { revision, bytes }, + other => other, + } + } + + fn junk(revision: u64, bytes: &[u8]) -> Read { + Read::Document { + revision, + bytes: bytes.to_vec(), + } + } + + #[test] + fn with_no_document_the_router_runs_the_builtins() { + let mut f = Flows::builtins(); + assert_eq!(ids(&f), ["assigned", "unassigned", "closed"]); + f.apply(Read::Absent); + assert_eq!(ids(&f), ["assigned", "unassigned", "closed"]); + // An empty document is the same. + f.apply(at(1, doc(&[], &[]))); + assert_eq!(ids(&f), ["assigned", "unassigned", "closed"]); + } + + #[test] + fn a_document_switches_builtins_off_and_adds_custom_flows() { + let mut f = Flows::builtins(); + f.apply(at(1, doc(&["unassigned"], &["triage"]))); + assert_eq!(ids(&f), ["assigned", "closed", "triage"]); + } + + #[test] + fn a_document_it_cannot_read_leaves_the_flows_as_they_were() { + let mut f = Flows::builtins(); + f.apply(at(1, doc(&["closed"], &["triage"]))); + let before = ids(&f).join(","); + assert_eq!(before, "assigned,unassigned,triage"); + for bad in [ + junk(2, b"not json"), + junk(3, br#"{"version":2,"flows":[]}"#), + junk(4, br#"{"flows":[{"id":"x"}]}"#), + Read::Failed("timeout".to_owned()), + ] { + f.apply(bad); + assert_eq!(ids(&f).join(","), before); + } + // The next good revision replaces them: the control that the bad + // ones were kept out by being bad, not by the loader being stuck. + f.apply(at(5, doc(&[], &[]))); + assert_eq!(ids(&f), ["assigned", "unassigned", "closed"]); + } + + #[test] + fn a_bad_first_document_is_the_builtins_not_nothing() { + let mut f = Flows::builtins(); + f.apply(junk(1, br#"{"version":9}"#)); + assert_eq!(ids(&f), ["assigned", "unassigned", "closed"]); + } + + #[test] + fn a_deleted_document_goes_back_to_the_builtins() { + let mut f = Flows::builtins(); + f.apply(at(1, doc(&["closed"], &["triage"]))); + f.apply(Read::Absent); + assert_eq!(ids(&f), ["assigned", "unassigned", "closed"]); + } + + #[test] + fn a_flow_that_is_not_valid_is_left_out_and_the_rest_run() { + let mut f = Flows::builtins(); + let mut broken = notify_flows::builtins().remove(0); + broken.id = "broken".to_owned(); + broken.edges.clear(); + let d = Document { + version: VERSION, + disabled_builtins: Vec::new(), + flows: vec![broken], + }; + f.apply(Read::Document { + revision: 1, + bytes: serde_json::to_vec(&d).unwrap(), + }); + assert_eq!(ids(&f), ["assigned", "unassigned", "closed"]); + } +} diff --git a/swarm-wire/src/lib.rs b/swarm-wire/src/lib.rs index 2b236b57..3d0c9b46 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -18,6 +18,7 @@ pub mod builder; pub mod claude_login; pub mod forge_events; pub mod notices; +pub mod notify_flows; pub mod notify_router; pub mod review_classifier; pub mod router_todo; diff --git a/swarm-wire/src/notify_flows.rs b/swarm-wire/src/notify_flows.rs new file mode 100644 index 00000000..60505666 --- /dev/null +++ b/swarm-wire/src/notify_flows.rs @@ -0,0 +1,774 @@ +//! The notification router's flows: the stored document, the flow type both +//! the built-in and the custom flows are, and the checks a flow passes. +//! +//! A flow is a graph of nodes joined by edges, rooted at one `source` node. +//! A `match` node passes when every one of its conditions holds, and then +//! each of its children runs. A `todo` node is a leaf and emits one todo op +//! per recipient. Flows are trees: one source, one parent for every other +//! node, every node reachable from the source ([`Flow::validate`]). +//! +//! The document under [`KEY`] in [`BUCKET`] holds the ids of the built-in +//! flows that are switched off and the custom flows. The built-ins +//! ([`builtins`]) are defined here, in code, so a build that adds one reaches +//! every swarm without a migration. swarm-controller is the only writer; the +//! router reads it. + +use std::collections::BTreeSet; + +use serde::{Deserialize, Serialize}; + +use crate::router_todo::Op; + +/// The bucket holding the document. Only the controller writes it. +pub const BUCKET: &str = "notify-flows"; + +/// The one key in [`BUCKET`]. +pub const KEY: &str = "all"; + +/// The document layout this build reads and writes. +pub const VERSION: u32 = 1; + +/// The most nodes one flow may have. +pub const MAX_NODES: usize = 64; + +/// The deepest a node may sit below the source. +pub const MAX_DEPTH: usize = 16; + +/// The most custom flows the document may hold. +pub const MAX_FLOWS: usize = 100; + +/// The longest a flow's name, or a todo's key or text, may be. +pub const MAX_TEXT: usize = 500; + +/// The forge event types a source node can start from: the ones +/// swarm-controller's hooks deliver to the events stream. +pub const SOURCE_EVENTS: &[&str] = &["issues", "issue_assign", "pull_request", "pull_request_assign"]; + +/// The fields of an event a condition or a recipient can name. +pub const FIELDS: &[&str] = &[ + "event", + "action", + "kind", + "repo", + "number", + "title", + "url", + "sender", + "assignees", + "assignees_added", + "assignees_removed", + "todo_holders", +]; + +/// The value under [`KEY`]. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct Document { + /// [`VERSION`] when written. A reader that does not know the number + /// leaves the document alone. + #[serde(default = "version")] + pub version: u32, + /// The ids of the built-in flows that are switched off. + #[serde(default)] + pub disabled_builtins: Vec, + /// The flows the operator defined. + #[serde(default)] + pub flows: Vec, +} + +fn version() -> u32 { + VERSION +} + +impl Default for Document { + /// No built-in switched off and no custom flow, at [`VERSION`]. + fn default() -> Self { + Self { + version: VERSION, + disabled_builtins: Vec::new(), + flows: Vec::new(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct Flow { + /// Unique among the built-in and custom flows. Names the flow in the + /// todo's source, `router:`. + pub id: String, + pub name: String, + /// A flow switched off is kept and runs on no event. Built-ins ignore it + /// and are switched off through [`Document::disabled_builtins`]. + #[serde(default = "on")] + pub enabled: bool, + pub nodes: Vec, + pub edges: Vec, +} + +fn on() -> bool { + true +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct Node { + /// Unique within its flow. + pub id: String, + #[serde(flatten)] + pub kind: NodeKind, +} + +/// A directed edge, parent to child. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct Edge { + pub from: String, + pub to: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum NodeKind { + /// The root: the forge event types the flow starts from, from + /// [`SOURCE_EVENTS`]. + Source { events: Vec }, + /// Passes when every condition holds. + Match { when: Vec }, + /// Emits a todo op to each recipient. + Todo(TodoSink), +} + +/// `field` equals `value`. A list field holds when one of its items does. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct Condition { + /// One of [`FIELDS`]. + pub field: String, + pub equals: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct TodoSink { + pub to: Recipients, + #[cfg_attr(feature = "schema", schema(value_type = String))] + pub op: Op, + /// The todo key, with `{field}` placeholders. + pub key: String, + /// For a clear: `key` is a prefix. + #[serde(default)] + pub prefix: bool, + /// The todo text, with `{field}` placeholders. Empty for a clear. + #[serde(default)] + pub text: String, +} + +/// Who a todo op goes to. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +#[serde(rename_all = "snake_case")] +pub enum Recipients { + /// The agents the named event field holds, one of [`FIELDS`]. + Field(String), + /// This agent, whatever the event. + Agent(String), +} + +/// Why [`Document::decode`] could not read a document. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DecodeError { + /// A version other than [`VERSION`], such as a newer build's. + UnknownVersion(String), + /// Not JSON, or not the shape of a [`Document`]. + Malformed(String), +} + +impl std::fmt::Display for DecodeError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::UnknownVersion(v) => { + write!(f, "version {v}, and this build reads version {VERSION}") + } + Self::Malformed(why) => f.write_str(why), + } + } +} + +impl Document { + /// `bytes`, the stored document, or why it cannot be read. + /// + /// # Errors + /// + /// Not the expected shape, or a [`Document::version`] this build does not + /// know. The reader keeps what it had. + pub fn decode(bytes: &[u8]) -> Result { + let value: serde_json::Value = + serde_json::from_slice(bytes).map_err(|e| DecodeError::Malformed(format!("not JSON: {e}")))?; + if let Some(found) = value.get("version") + && found != VERSION + { + return Err(DecodeError::UnknownVersion(found.to_string())); + } + serde_json::from_value(value) + .map_err(|e| DecodeError::Malformed(format!("not a flows document: {e}"))) + } + + /// What is wrong with the document, if anything: the check a save passes. + /// + /// # Errors + /// + /// The first problem found, naming the flow it is in. + pub fn check(&self) -> Result<(), String> { + if self.version != VERSION { + return Err(format!("version {} is not {VERSION}", self.version)); + } + if self.flows.len() > MAX_FLOWS { + return Err(format!( + "{} flows, more than {MAX_FLOWS}", + self.flows.len() + )); + } + let builtin = builtins(); + for id in &self.disabled_builtins { + if !builtin.iter().any(|b| b.id == *id) { + return Err(format!("disabled_builtins names {id}, which is no built-in flow")); + } + } + let mut ids: BTreeSet<&str> = builtin.iter().map(|b| b.id.as_str()).collect(); + for flow in &self.flows { + if !ids.insert(flow.id.as_str()) { + return Err(format!( + "flow id {} is used twice, or is a built-in flow's", + flow.id + )); + } + flow.validate().map_err(|why| format!("flow {}: {why}", flow.id))?; + } + Ok(()) + } + + /// The flows to run: the built-ins that are not switched off, then the + /// enabled custom flows that pass [`Flow::validate`] with an id of their + /// own. With the reason for each custom flow left out. + #[must_use] + pub fn active(&self) -> (Vec, Vec) { + let mut flows: Vec = builtins() + .into_iter() + .filter(|b| !self.disabled_builtins.contains(&b.id)) + .collect(); + let mut skipped = Vec::new(); + for flow in self.flows.iter().filter(|f| f.enabled) { + if flows.iter().any(|f| f.id == flow.id) { + skipped.push(format!("flow {}: its id is already in use", flow.id)); + } else if let Err(why) = flow.validate() { + skipped.push(format!("flow {}: {why}", flow.id)); + } else { + flows.push(flow.clone()); + } + } + (flows, skipped) + } +} + +/// Whether `id` can name a flow or a node: it ends up in a message id and in +/// a todo's source. +fn is_id(id: &str) -> bool { + !id.is_empty() + && id.len() <= 64 + && id + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') +} + +fn is_field(name: &str) -> bool { + FIELDS.contains(&name) +} + +impl Flow { + /// Check the flow is a tree within the caps, and that every node reads + /// only what exists. + /// + /// # Errors + /// + /// Says why not. + pub fn validate(&self) -> Result<(), String> { + if !is_id(&self.id) { + return Err(format!( + "id {:?} is not 1 to 64 letters, digits, - or _", + self.id + )); + } + if self.name.trim().is_empty() || self.name.len() > MAX_TEXT { + return Err(format!("name is empty or longer than {MAX_TEXT} bytes")); + } + if self.nodes.len() > MAX_NODES { + return Err(format!("{} nodes, more than {MAX_NODES}", self.nodes.len())); + } + let mut ids = BTreeSet::new(); + for node in &self.nodes { + if !is_id(&node.id) { + return Err(format!( + "node id {:?} is not 1 to 64 letters, digits, - or _", + node.id + )); + } + if !ids.insert(node.id.as_str()) { + return Err(format!("node id {} is used twice", node.id)); + } + node.kind + .validate() + .map_err(|why| format!("node {}: {why}", node.id))?; + } + for edge in &self.edges { + for end in [&edge.from, &edge.to] { + if !ids.contains(end.as_str()) { + return Err(format!("an edge names {end}, which is no node")); + } + } + } + let roots: Vec<&Node> = self + .nodes + .iter() + .filter(|n| matches!(n.kind, NodeKind::Source { .. })) + .collect(); + let [root] = roots.as_slice() else { + return Err(format!("{} source nodes, not one", roots.len())); + }; + for node in &self.nodes { + if matches!(node.kind, NodeKind::Todo(_)) && !self.children(&node.id).is_empty() { + return Err(format!("todo {} has children", node.id)); + } + } + // From the one root, every node reached exactly once: a node reached + // twice has a second parent or sits on a cycle, and one not reached is + // cut off. + let mut reached = 0; + let mut stack = vec![(root.id.as_str(), 0)]; + while let Some((id, depth)) = stack.pop() { + if depth > MAX_DEPTH { + return Err(format!("node {id} is deeper than {MAX_DEPTH}")); + } + reached += 1; + if reached > self.nodes.len() { + return Err( + "a node is reached twice: it has two parents or sits on a cycle".to_owned(), + ); + } + stack.extend(self.children(id).into_iter().map(|c| (c, depth + 1))); + } + if reached != self.nodes.len() { + return Err(format!( + "{} of {} nodes are not reachable from the source", + self.nodes.len() - reached, + self.nodes.len() + )); + } + Ok(()) + } + + /// The ids of the children of node `id`. + #[must_use] + pub fn children(&self, id: &str) -> Vec<&str> { + self.edges + .iter() + .filter(|e| e.from == id) + .map(|e| e.to.as_str()) + .collect() + } + + /// The node with `id`. + #[must_use] + pub fn node(&self, id: &str) -> Option<&Node> { + self.nodes.iter().find(|n| n.id == id) + } +} + +impl NodeKind { + fn validate(&self) -> Result<(), String> { + match self { + Self::Source { events } => { + if events.is_empty() { + return Err("a source with no events".to_owned()); + } + if let Some(bad) = events.iter().find(|e| !SOURCE_EVENTS.contains(&e.as_str())) { + return Err(format!( + "event {bad:?} is not one of {}", + SOURCE_EVENTS.join(", ") + )); + } + } + Self::Match { when } => { + if let Some(bad) = when.iter().find(|c| !is_field(&c.field)) { + return Err(format!("condition on {:?}, which is no event field", bad.field)); + } + } + Self::Todo(sink) => sink.validate()?, + } + Ok(()) + } +} + +impl TodoSink { + fn validate(&self) -> Result<(), String> { + match &self.to { + Recipients::Field(f) if !is_field(f) => { + return Err(format!("recipients from {f:?}, which is no event field")); + } + Recipients::Agent(a) if !is_id(a) => { + return Err(format!("recipient {a:?} is not an agent name")); + } + _ => {} + } + if self.key.is_empty() || self.key.len() > MAX_TEXT { + return Err(format!("key is empty or longer than {MAX_TEXT} bytes")); + } + if self.text.len() > MAX_TEXT { + return Err(format!("text is longer than {MAX_TEXT} bytes")); + } + match self.op { + Op::Upsert if self.text.trim().is_empty() => Err("an upsert with no text".to_owned()), + Op::Upsert if self.prefix => Err("prefix is for a clear".to_owned()), + Op::Clear if !self.text.is_empty() => Err("a clear takes no text".to_owned()), + _ => Ok(()), + } + } +} + +/// The flows every swarm ships with. +/// +/// - `assigned`: a newly assigned agent gets a todo for the issue or PR. +/// - `unassigned`: an agent taken off it gets that todo rewritten to say so. +/// Both write the same key, so the second replaces the first. +/// - `closed`: closing it clears every router todo the router queued for it. +#[must_use] +pub fn builtins() -> Vec { + let assignment_key = "forge:{repo}#{number}:assignment"; + vec![ + chain( + "assigned", + &["issue_assign", "pull_request_assign"], + "assigned", + TodoSink { + to: Recipients::Field("assignees_added".to_owned()), + op: Op::Upsert, + key: assignment_key.to_owned(), + prefix: false, + text: "you are assigned to {kind} {repo}#{number}: {title} ({url})".to_owned(), + }, + ), + chain( + "unassigned", + &["issue_assign", "pull_request_assign"], + "unassigned", + TodoSink { + to: Recipients::Field("assignees_removed".to_owned()), + op: Op::Upsert, + key: assignment_key.to_owned(), + prefix: false, + text: "you are not assigned to {kind} {repo}#{number} anymore: {title} ({url})" + .to_owned(), + }, + ), + chain( + "closed", + &["issues", "pull_request"], + "closed", + TodoSink { + to: Recipients::Field("todo_holders".to_owned()), + op: Op::Clear, + key: "forge:{repo}#{number}:".to_owned(), + prefix: true, + text: String::new(), + }, + ), + ] +} + +/// source(`events`) → match(`action` = `action`) → `sink`. +fn chain(id: &str, events: &[&str], action: &str, sink: TodoSink) -> Flow { + let node = |id: &str, kind| Node { + id: id.to_owned(), + kind, + }; + let edge = |from: &str, to: &str| Edge { + from: from.to_owned(), + to: to.to_owned(), + }; + Flow { + id: id.to_owned(), + name: id.to_owned(), + enabled: true, + nodes: vec![ + node( + "source", + NodeKind::Source { + events: events.iter().map(|e| (*e).to_owned()).collect(), + }, + ), + node( + "match", + NodeKind::Match { + when: vec![Condition { + field: "action".to_owned(), + equals: action.to_owned(), + }], + }, + ), + node("sink", NodeKind::Todo(sink)), + ], + edges: vec![edge("source", "match"), edge("match", "sink")], + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn edge(from: &str, to: &str) -> Edge { + Edge { + from: from.to_owned(), + to: to.to_owned(), + } + } + + fn builtin(id: &str) -> Flow { + builtins().into_iter().find(|f| f.id == id).expect("a built-in") + } + + /// A custom flow: opened issue or PR → a todo for `triage`. + fn custom(id: &str) -> Flow { + let mut f = builtin("assigned"); + f.id = id.to_owned(); + f.name = "new to triage".to_owned(); + f.nodes[2].kind = NodeKind::Todo(TodoSink { + to: Recipients::Agent("triage".to_owned()), + op: Op::Upsert, + key: "forge:{repo}#{number}:triage".to_owned(), + prefix: false, + text: "triage {repo}#{number}".to_owned(), + }); + f + } + + #[test] + fn every_builtin_is_a_valid_tree() { + for f in builtins() { + assert_eq!(f.validate(), Ok(()), "{}", f.id); + } + } + + #[test] + fn a_flow_that_is_not_a_tree_is_refused() { + let mut cycle = builtin("assigned"); + cycle.edges[0] = edge("sink", "match"); + cycle.edges.push(edge("match", "sink")); + assert!(cycle.validate().is_err(), "a cycle"); + + let mut orphan = builtin("assigned"); + orphan.edges.pop(); + assert!(orphan.validate().is_err(), "an unreachable sink"); + + let mut two_parents = builtin("assigned"); + two_parents.edges.push(edge("source", "sink")); + assert!(two_parents.validate().is_err(), "a sink with two parents"); + + let mut sink_child = builtin("assigned"); + sink_child.nodes.push(Node { + id: "m2".to_owned(), + kind: NodeKind::Match { when: Vec::new() }, + }); + sink_child.edges.push(edge("sink", "m2")); + assert!(sink_child.validate().is_err(), "a sink with a child"); + + let mut dangling = builtin("assigned"); + dangling.edges.push(edge("match", "nowhere")); + assert!(dangling.validate().is_err(), "an edge to no node"); + + let mut two_sources = builtin("assigned"); + two_sources.nodes[1].kind = NodeKind::Source { + events: vec!["issues".to_owned()], + }; + assert!(two_sources.validate().is_err(), "two sources"); + + let mut twice = builtin("assigned"); + twice.nodes[1].id = "sink".to_owned(); + assert!(twice.validate().is_err(), "a node id used twice"); + } + + #[test] + fn the_caps_hold_at_the_limit_and_refuse_one_past_it() { + // A chain source → m0 → m1 → … of `n` matches under the source. + let chain_of = |n: usize| { + let mut f = builtin("assigned"); + f.nodes.truncate(1); + f.edges.clear(); + let mut parent = "source".to_owned(); + for i in 0..n { + let id = format!("m{i}"); + f.nodes.push(Node { + id: id.clone(), + kind: NodeKind::Match { when: Vec::new() }, + }); + f.edges.push(edge(&parent, &id)); + parent = id; + } + f + }; + assert_eq!(chain_of(MAX_DEPTH).validate(), Ok(())); + assert!(chain_of(MAX_DEPTH + 1).validate().is_err(), "depth"); + let mut wide = chain_of(0); + for i in 0..MAX_NODES - 1 { + wide.nodes.push(Node { + id: format!("m{i}"), + kind: NodeKind::Match { when: Vec::new() }, + }); + wide.edges.push(edge("source", &format!("m{i}"))); + } + assert_eq!(wide.validate(), Ok(())); + wide.nodes.push(Node { + id: "extra".to_owned(), + kind: NodeKind::Match { when: Vec::new() }, + }); + wide.edges.push(edge("source", "extra")); + assert!(wide.validate().is_err(), "node count"); + } + + #[test] + fn a_node_that_reads_what_does_not_exist_is_refused() { + let mut event = builtin("assigned"); + event.nodes[0].kind = NodeKind::Source { + events: vec!["pull_request_sync".to_owned()], + }; + assert!(event.validate().unwrap_err().contains("pull_request_sync")); + + let mut field = builtin("assigned"); + field.nodes[1].kind = NodeKind::Match { + when: vec![Condition { + field: "actoin".to_owned(), + equals: "closed".to_owned(), + }], + }; + assert!(field.validate().unwrap_err().contains("actoin")); + + let mut to = custom("c"); + to.nodes[2].kind = NodeKind::Todo(TodoSink { + to: Recipients::Field("nobody".to_owned()), + op: Op::Upsert, + key: "k".to_owned(), + prefix: false, + text: "t".to_owned(), + }); + assert!(to.validate().unwrap_err().contains("nobody")); + // Control: a literal agent and the sources above spelled right pass. + assert_eq!(custom("c").validate(), Ok(())); + } + + #[test] + fn a_sink_must_say_what_its_op_needs() { + let sink = |op, prefix, text: &str| { + let mut f = custom("c"); + f.nodes[2].kind = NodeKind::Todo(TodoSink { + to: Recipients::Agent("triage".to_owned()), + op, + key: "k".to_owned(), + prefix, + text: text.to_owned(), + }); + f.validate() + }; + assert_eq!(sink(Op::Upsert, false, "t"), Ok(())); + assert_eq!(sink(Op::Clear, true, ""), Ok(())); + assert!(sink(Op::Upsert, false, " ").is_err(), "no text"); + assert!(sink(Op::Upsert, true, "t").is_err(), "prefix on an upsert"); + assert!(sink(Op::Clear, false, "t").is_err(), "text on a clear"); + } + + #[test] + fn a_document_names_only_real_builtins_and_unique_flow_ids() { + let doc = |disabled: &[&str], flows: Vec| Document { + version: VERSION, + disabled_builtins: disabled.iter().map(|s| (*s).to_owned()).collect(), + flows, + }; + assert_eq!(doc(&["assigned"], vec![custom("triage")]).check(), Ok(())); + assert!(doc(&["nope"], vec![]).check().unwrap_err().contains("nope")); + assert!( + doc(&[], vec![custom("assigned")]).check().is_err(), + "a built-in's id" + ); + assert!( + doc(&[], vec![custom("a"), custom("a")]).check().is_err(), + "a repeated id" + ); + let mut bad = custom("bad"); + bad.edges.pop(); + assert!(doc(&[], vec![bad]).check().unwrap_err().starts_with("flow bad:")); + let many = (0..=MAX_FLOWS).map(|i| custom(&format!("f{i}"))).collect(); + assert!(doc(&[], many).check().is_err(), "flow count"); + let mut old = doc(&[], vec![]); + old.version = 2; + assert!(old.check().is_err(), "version"); + } + + #[test] + fn active_is_the_builtins_minus_the_disabled_plus_the_enabled_valid_custom() { + let ids = |d: &Document| d.active().0.into_iter().map(|f| f.id).collect::>(); + assert_eq!( + ids(&Document::default()), + ["assigned", "unassigned", "closed"], + "an empty document is the built-ins" + ); + let mut off = custom("off"); + off.enabled = false; + let mut broken = custom("broken"); + broken.edges.clear(); + let doc = Document { + version: VERSION, + disabled_builtins: vec!["unassigned".to_owned(), "gone".to_owned()], + flows: vec![custom("triage"), off, broken, custom("assigned")], + }; + assert_eq!(ids(&doc), ["assigned", "closed", "triage"]); + let (_, skipped) = doc.active(); + assert_eq!(skipped.len(), 2, "{skipped:?}"); + assert!(skipped.iter().any(|s| s.starts_with("flow broken:"))); + assert!(skipped.iter().any(|s| s.contains("assigned"))); + } + + #[test] + fn a_document_reads_back_and_a_foreign_one_is_refused() { + let doc = Document { + version: VERSION, + disabled_builtins: vec!["closed".to_owned()], + flows: vec![custom("triage")], + }; + let bytes = serde_json::to_vec(&doc).unwrap(); + assert_eq!(Document::decode(&bytes), Ok(doc)); + assert_eq!(Document::decode(b"{}"), Ok(Document::default())); + assert_eq!( + Document::decode(br#"{"version":2}"#), + Err(DecodeError::UnknownVersion("2".to_owned())) + ); + assert!(Document::decode(b"nope").is_err()); + assert!(Document::decode(br#"{"flows":[{"id":"x"}]}"#).is_err()); + } + + /// The wire shape the API and the stored document share. + #[test] + fn a_node_is_flat_with_its_kind_beside_its_id() { + let json = serde_json::to_value(&builtin("closed").nodes[2]).unwrap(); + assert_eq!( + json, + serde_json::json!({ + "id": "sink", + "kind": "todo", + "to": { "field": "todo_holders" }, + "op": "clear", + "key": "forge:{repo}#{number}:", + "prefix": true, + "text": "", + }) + ); + } +}