notify router: read custom flows and disabled built-ins from a stored document

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;

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<u64>,
+    /// The ids of the built-in flows that are switched off.
+    disabled_builtins: Vec<String>,
+    /// The built-in flows this build ships, all of them, switched off or not.
+    builtins: Vec<Flow>,
+    /// The custom flows.
+    flows: Vec<Flow>,
+}
+
+/// 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<u64>,
+    /// The ids of the built-in flows to switch off.
+    #[serde(default)]
+    disabled_builtins: Vec<String>,
+    /// The custom flows.
+    #[serde(default)]
+    flows: Vec<Flow>,
+}
+
+impl FlowsView {
+    fn of(doc: Document, revision: Option<u64>) -> 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<u64>)> {
+    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<Store> {
+    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<AppState>,
+) -> Result<Json<FlowsView>, 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<AppState>,
+    body: Result<Json<PutFlows>, JsonRejection>,
+) -> Result<Json<FlowsView>, 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(_)) = &current
+        {
+            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"
+    )))
+}

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