//! `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 std::collections::{BTreeMap, BTreeSet};

use swarm_queue_client::archetypes::{self, SourceRecord, flows_of};
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_or_default, 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`; `0` when
    /// none is stored.
    revision: u64,
    /// The ids of the built-in flows that are switched off, and the keys of the
    /// archetype flows that are.
    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 flows archetypes carry, each with the agents it runs for. They follow
    /// their archetype, so they are not in the document; a `PUT` switches one off
    /// with its `key` in `disabled_builtins`. An older controller's answer has
    /// none.
    archetype_flows: Vec<ArchetypeFlowView>,
    /// What this build accepts: source events, fields, node kinds and limits,
    /// for an editor to offer. An older controller's answer has none.
    schema: notify_flows::Schema,
}

/// One flow an archetype carries.
#[derive(Debug, Clone, PartialEq, Serialize, ToSchema)]
pub(crate) struct ArchetypeFlowView {
    /// The archetype's name.
    archetype: String,
    /// What names the flow in `disabled_builtins` (off for every member), in the
    /// router's log and in a todo's source. `<key>@<agent>` in
    /// `disabled_builtins` switches it off for one member.
    key: String,
    /// The flow as the archetype declares it, under its own `id`.
    flow: Flow,
    /// The agents of the archetype, by name.
    members: Vec<String>,
}

/// 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, `0` for "no document stored". The
    /// `PUT` is refused when the stored revision is another, which includes a
    /// document written since a client read `0`. A KV revision is never `0`.
    revision: 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: u64, archetype_flows: Vec<ArchetypeFlowView>) -> Self {
        Self {
            revision,
            disabled_builtins: doc.disabled_builtins,
            builtins: notify_flows::builtins(),
            flows: doc.flows,
            archetype_flows,
            schema: notify_flows::schema(),
        }
    }
}

/// What the key holds.
pub(crate) 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.
pub(crate) 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.
pub(crate) 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))
}

pub(crate) 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. Creates the `notify-flows` bucket when it is missing, as a
/// `PUT` does, so the router finds it empty rather than absent.
#[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.unwrap_or(0),
            archetype_flows(&client).await,
        ))),
        Stored::Document(doc) => Ok(Json(FlowsView::of(
            doc,
            revision.unwrap_or(0),
            archetype_flows(&client).await,
        ))),
        Stored::Unreadable(e) => Err(unreadable(&e)),
    }
}

/// The flows the archetypes carry, with their agents. A read that fails is
/// logged and leaves the list empty: the document is still served.
async fn archetype_flows(client: &async_nats::Client) -> Vec<ArchetypeFlowView> {
    let found = async {
        // The router's members: the archetype keys of the agents in the swarm.
        let roster = swarm_queue_client::agent_status::roster(client)
            .await
            .map_err(|e| anyhow::anyhow!("listing the agents: {e}"))?;
        let mut refs = crate::agent_archetype::all(client).await?;
        refs.retain(|agent, _| roster.contains(agent));
        let mut records = BTreeMap::new();
        if let Some(store) = archetypes::open(client).await? {
            let sources: BTreeSet<&str> = refs.values().map(|r| r.source.as_str()).collect();
            for source in sources {
                let (record, _, _) = read_or_default::<SourceRecord>(&store, source).await?;
                records.insert(source.to_owned(), record);
            }
        }
        anyhow::Ok(flows_of(&refs, &records).0)
    }
    .await;
    match found {
        Ok(found) => found
            .into_iter()
            .map(|f| ArchetypeFlowView {
                key: f.key(),
                archetype: f.archetype,
                flow: f.flow,
                members: f.members,
            })
            .collect(),
        Err(e) => {
            tracing::warn!(
                error = %format!("{e:#}"),
                "notify flows: reading the archetype flows failed; none shown"
            );
            Vec::new()
        }
    }
}

/// `disabled` less the switch-offs for `agent` alone (`archetype:…/…@agent`; no
/// built-in id has an `@`),
/// or `None` when it has none. The per-archetype keys stay.
fn without_agent_keys(disabled: &[String], agent: &str) -> Option<Vec<String>> {
    let suffix = format!("@{agent}");
    let kept: Vec<String> = disabled
        .iter()
        .filter(|k| !k.ends_with(&suffix))
        .cloned()
        .collect();
    (kept.len() != disabled.len()).then_some(kept)
}

/// Remove the keys that switch archetype flows off for `agent` from the stored
/// document, in the compare-and-set loop a `PUT` uses, so a concurrent edit
/// is not lost. Answers the keys removed. No write when there are none.
async fn clear_agent_keys(client: &async_nats::Client, agent: &str) -> anyhow::Result<Vec<String>> {
    let store = open(client).await?;
    for _ in 0..MAX_ATTEMPTS {
        let (current, revision) = stored(&store).await?;
        let Stored::Document(mut doc) = current else {
            return Ok(Vec::new());
        };
        let Some(kept) = without_agent_keys(&doc.disabled_builtins, agent) else {
            return Ok(Vec::new());
        };
        let removed: Vec<String> = doc
            .disabled_builtins
            .iter()
            .filter(|k| !kept.contains(k))
            .cloned()
            .collect();
        doc.disabled_builtins = kept;
        if write_revision(&store, notify_flows::KEY, &doc, revision)
            .await?
            .is_some()
        {
            return Ok(removed);
        }
    }
    anyhow::bail!("gave up after {MAX_ATTEMPTS} conflicting writes")
}

/// Clear a destroyed agent's per-agent switch-offs, so a new agent of that
/// name starts with every archetype flow on. A failure is logged with the keys
/// left behind and does not stop the destroy.
pub(crate) async fn clear_on_destroy(state: &AppState, hive: &str, agent: &str) {
    let cleared = async {
        let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?;
        clear_agent_keys(&client, agent).await
    }
    .await;
    match cleared {
        Ok(removed) if removed.is_empty() => {}
        Ok(removed) => {
            tracing::info!(
                hive,
                agent,
                ?removed,
                "destroyed agent's flow switch-offs cleared"
            );
        }
        Err(e) => tracing::error!(
            hive,
            agent,
            error = %format!("{e:#}"),
            "clearing this agent's flow switch-offs (`archetype:…@<agent>` in the flows \
             document's disabled list) FAILED; it was destroyed anyway, so a new agent of the \
             same name starts with those flows off. Declaring the agent destroyed again \
             re-runs this."
        ),
    }
}

pub(crate) fn unreadable(why: &DecodeError) -> ProblemDetails {
    error_problem(
        StatusCode::CONFLICT,
        &format!("the stored flows document is not one this build reads: {why}"),
    )
}

/// The document a body describes, or the first way it fails
/// [`Document::check`].
pub(crate) fn checked_document(
    disabled_builtins: Vec<String>,
    flows: Vec<Flow>,
) -> Result<Document, String> {
    let doc = Document {
        version: notify_flows::VERSION,
        disabled_builtins,
        flows,
    };
    doc.check().map(|()| doc)
}

/// 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 = "the stored document is not at `revision` (another write landed, or one now exists where the client read none), 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 = checked_document(put.disabled_builtins, put.flows)
        .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 != revision.unwrap_or(0) {
            return Err(error_problem(
                StatusCode::CONFLICT,
                &format!(
                    "the flows are not at revision {}; read them again",
                    put.revision
                ),
            ));
        }
        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,
                written,
                archetype_flows(&client).await,
            )));
        }
    }
    Err(queue_problem(&anyhow::anyhow!(
        "gave up writing the notification flows after {MAX_ATTEMPTS} conflicting writes"
    )))
}

#[cfg(test)]
mod tests {
    use swarm_wire::notify_flows::{Edge, Node, NodeKind, Recipients, TodoSink};
    use swarm_wire::router_todo::Op;

    use super::*;
    use crate::test_nats::TestNats;

    fn custom(id: &str) -> Flow {
        Flow {
            id: id.to_owned(),
            name: "new to triage".to_owned(),
            enabled: true,
            nodes: vec![
                Node {
                    id: "source".to_owned(),
                    kind: NodeKind::Source {
                        events: vec!["issues".to_owned()],
                    },
                },
                Node {
                    id: "sink".to_owned(),
                    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(),
                    }),
                },
            ],
            edges: vec![Edge {
                from: "source".to_owned(),
                to: "sink".to_owned(),
            }],
        }
    }

    fn body(revision: u64, disabled: &[&str], flows: Vec<Flow>) -> Json<PutFlows> {
        Json(PutFlows {
            revision,
            disabled_builtins: disabled.iter().map(|s| (*s).to_owned()).collect(),
            flows,
        })
    }

    #[tokio::test]
    async fn the_flows_are_read_replaced_and_checked_on_save() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::state(&client);

        // Nothing stored: the built-ins alone, none off, no revision.
        let Json(view) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(view.revision, 0);
        assert!(view.disabled_builtins.is_empty() && view.flows.is_empty());
        assert_eq!(view.builtins.len(), notify_flows::builtins().len());
        assert!(view.builtins.iter().any(|f| f.id == "head-classifier"));
        let json = serde_json::to_value(&view).expect("serializes");
        assert_eq!(json["schema"]["limits"]["max_nodes"], 64);
        assert_eq!(json["schema"]["node_kinds"][0]["kind"], "source");
        assert_eq!(view.schema, notify_flows::schema());

        let Json(saved) = put_flows(
            State(state.clone()),
            Ok(body(0, &["unassigned"], vec![custom("triage")])),
        )
        .await
        .expect("put");
        let revision = saved.revision;
        assert_ne!(revision, 0);
        let Json(read) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(read, saved);
        assert_eq!(read.disabled_builtins, ["unassigned"]);
        assert_eq!(read.flows[0].id, "triage");

        // The stored bytes are the document the router decodes.
        let store = open(&client).await.expect("bucket");
        let bytes = store
            .get(notify_flows::KEY)
            .await
            .expect("read")
            .expect("stored");
        let doc = Document::decode(&bytes).expect("the router reads it");
        assert_eq!(
            doc.active().0.len(),
            notify_flows::builtins().len(),
            "the built-ins less the switched-off one, and the custom flow"
        );

        // A refused save changes nothing: an orphan node, and a built-in that does not exist.
        let mut orphan = custom("orphan");
        orphan.edges.clear();
        let refused = put_flows(State(state.clone()), Ok(body(revision, &[], vec![orphan])))
            .await
            .expect_err("an orphan node");
        assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{refused:?}");
        let unknown = put_flows(State(state.clone()), Ok(body(revision, &["nope"], vec![])))
            .await
            .expect_err("no such built-in");
        assert_eq!(unknown.status, Some(StatusCode::BAD_REQUEST), "{unknown:?}");
        let Json(after) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(after, saved);

        // A stale revision is refused; the current one lands.
        let outdated = put_flows(State(state.clone()), Ok(body(revision + 100, &[], vec![])))
            .await
            .expect_err("stale");
        assert_eq!(outdated.status, Some(StatusCode::CONFLICT), "{outdated:?}");
        let Json(next) = put_flows(State(state.clone()), Ok(body(revision, &[], vec![])))
            .await
            .expect("current revision");
        assert!(next.revision > revision);
        assert!(next.flows.is_empty());
    }

    /// A document of a version this build does not know is neither served
    /// nor replaced, and a garbled one is replaced.
    #[tokio::test]
    async fn a_document_of_another_version_is_kept_and_a_garbled_one_is_replaced() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::state(&client);
        let store = open(&client).await.expect("bucket");

        store
            .put(
                notify_flows::KEY,
                br#"{"version":3,"flows":[]}"#.to_vec().into(),
            )
            .await
            .expect("seed");
        let got = get_flows(State(state.clone()))
            .await
            .expect_err("version 3");
        assert_eq!(got.status, Some(StatusCode::CONFLICT), "{got:?}");
        let refused = put_flows(State(state.clone()), Ok(body(0, &[], vec![])))
            .await
            .expect_err("version 3 is kept");
        assert_eq!(refused.status, Some(StatusCode::CONFLICT), "{refused:?}");
        let kept = store
            .get(notify_flows::KEY)
            .await
            .expect("read")
            .expect("stored");
        assert_eq!(&kept[..], br#"{"version":3,"flows":[]}"#);

        // Control: garbage that names no version is replaced by a save.
        let garbage = store
            .put(notify_flows::KEY, b"garbage".to_vec().into())
            .await
            .expect("seed");
        let _ = put_flows(State(state.clone()), Ok(body(garbage, &["closed"], vec![])))
            .await
            .expect("replaces garbage");
        let Json(view) = get_flows(State(state)).await.expect("readable again");
        assert_eq!(view.disabled_builtins, ["closed"]);
    }

    /// A page that read no document cannot erase one written since: its
    /// `revision: 0` means "create", and the document exists by then.
    #[tokio::test]
    async fn a_page_that_read_nothing_cannot_erase_a_flow_written_after() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::state(&client);

        let Json(page) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(page.revision, 0);
        // The API writes a custom flow while the page is open.
        let Json(api) = put_flows(
            State(state.clone()),
            Ok(body(0, &[], vec![custom("triage")])),
        )
        .await
        .expect("the API's write");
        // The page's first switch carries what it read.
        let erased = put_flows(
            State(state.clone()),
            Ok(body(page.revision, &["closed"], vec![])),
        )
        .await
        .expect_err("the page read nothing, and a document exists");
        assert_eq!(erased.status, Some(StatusCode::CONFLICT), "{erased:?}");
        let Json(after) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(after, api);
        assert_eq!(after.flows[0].id, "triage");

        // Control: re-read, the same switch lands and keeps the flow.
        let Json(switched) = put_flows(
            State(state),
            Ok(body(after.revision, &["closed"], after.flows.clone())),
        )
        .await
        .expect("the switch with the revision it read");
        assert_eq!(switched.disabled_builtins, ["closed"]);
        assert_eq!(switched.flows[0].id, "triage");
    }

    #[tokio::test]
    async fn a_template_naming_no_event_field_is_refused_on_save() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::state(&client);

        let mut typo = custom("typo");
        if let NodeKind::Todo(sink) = &mut typo.nodes[1].kind {
            sink.key = "forge:{{ repo }}#{{ numbr }}:triage".to_owned();
        }
        let refused = put_flows(State(state.clone()), Ok(body(0, &[], vec![typo])))
            .await
            .expect_err("numbr is no field");
        assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{refused:?}");
        // Control: the same flow spelled right is stored.
        let _ = put_flows(State(state), Ok(body(0, &[], vec![custom("typo")])))
            .await
            .expect("{number} is a field");
    }

    /// The archetypes' flows are listed with their agents; their keys go in the
    /// disabled list, a key for a flow that is gone is kept, and a key of no
    /// valid shape is refused.
    #[tokio::test]
    async fn archetype_flows_are_listed_and_their_keys_are_saved() {
        use swarm_queue_client::agent_archetype::ArchetypeRef;
        use swarm_queue_client::archetypes::{ArchetypeMeta, BUILTIN_SOURCE, SourceRecord};

        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::state(&client);

        let mut members = custom("new-issues");
        members.nodes[1].kind = NodeKind::Todo(TodoSink {
            to: Recipients::Members,
            op: Op::Upsert,
            key: "forge:{{ repo }}#{{ number }}:triage".to_owned(),
            prefix: false,
            text: "triage {{ repo }}#{{ number }}".to_owned(),
        });
        let record = SourceRecord {
            source: BUILTIN_SOURCE.to_owned(),
            archetypes: [
                (
                    "triage".to_owned(),
                    ArchetypeMeta {
                        description: "d".to_owned(),
                        flows: vec![members.clone()],
                        ..ArchetypeMeta::default()
                    },
                ),
                (
                    "reviewer".to_owned(),
                    ArchetypeMeta {
                        description: "d".to_owned(),
                        ..ArchetypeMeta::default()
                    },
                ),
            ]
            .into(),
            refused: BTreeMap::default(),
        };
        let records = swarm_queue_client::archetypes::open_or_create(&client)
            .await
            .expect("records");
        records
            .put(BUILTIN_SOURCE, serde_json::to_vec(&record).unwrap().into())
            .await
            .expect("put the record");
        let choices = swarm_queue_client::agent_archetype::open_or_create(&client)
            .await
            .expect("choices");
        for (agent, archetype) in [
            ("alice", "triage"),
            ("bob", "triage"),
            ("carol", "reviewer"),
            // In the bucket but not in the swarm: the router sends it nothing.
            ("dave", "triage"),
        ] {
            let choice = ArchetypeRef {
                source: BUILTIN_SOURCE.to_owned(),
                name: archetype.to_owned(),
            };
            choices
                .put(agent, serde_json::to_vec(&choice).unwrap().into())
                .await
                .expect("put the choice");
        }

        let status = swarm_queue_client::agent_status::open_or_create(&client)
            .await
            .expect("status");
        for agent in ["alice", "bob", "carol"] {
            status
                .put(format!("hive.{agent}"), "{}".into())
                .await
                .expect("put the status");
        }

        let Json(view) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(view.archetype_flows.len(), 1, "{:?}", view.archetype_flows);
        let listed = &view.archetype_flows[0];
        assert_eq!(listed.key, "archetype:triage/new-issues");
        assert_eq!(listed.archetype, "triage");
        assert_eq!(listed.flow, members);
        assert_eq!(listed.members, ["alice", "bob"]);

        // Off for everyone, off for one agent, and a key of a flow that is gone.
        let keys = [
            "archetype:triage/new-issues",
            "archetype:triage/new-issues@alice",
            "archetype:gone/flow",
        ];
        let Json(saved) = put_flows(State(state.clone()), Ok(body(0, &keys, vec![])))
            .await
            .expect("the keys are saved");
        assert_eq!(saved.disabled_builtins, keys);
        assert_eq!(
            saved.archetype_flows.len(),
            1,
            "a PUT answers with them too"
        );
        for bad in ["archetype:triage", "archetype:/x", "archetype:a/b@", "nope"] {
            let refused = put_flows(
                State(state.clone()),
                Ok(body(saved.revision, &[bad], vec![])),
            )
            .await
            .expect_err(bad);
            assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{bad}");
        }
    }

    /// Destroying an agent removes its own switch-offs from the document and
    /// nothing else; a document without any is not written.
    #[tokio::test]
    async fn a_destroyed_agents_switch_offs_are_cleared_and_nothing_else() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::state(&client);

        let keys = [
            "archetype:a/f",
            "archetype:a/f@alice",
            "archetype:a/f@bob",
            "archetype:a/g@alice",
            "closed",
        ];
        let Json(saved) = put_flows(State(state.clone()), Ok(body(0, &keys, vec![])))
            .await
            .expect("put");

        let removed = clear_agent_keys(&client, "alice").await.expect("clear");
        assert_eq!(removed, ["archetype:a/f@alice", "archetype:a/g@alice"]);
        let Json(after) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(
            after.disabled_builtins,
            ["archetype:a/f", "archetype:a/f@bob", "closed"]
        );
        assert!(after.revision > saved.revision);

        // Nothing of alice's left: no write.
        assert!(clear_agent_keys(&client, "alice").await.unwrap().is_empty());
        let Json(same) = get_flows(State(state.clone())).await.expect("get");
        assert_eq!(same.revision, after.revision);
        // A name that is the end of another's is not it.
        assert!(clear_agent_keys(&client, "ob").await.unwrap().is_empty());
        let Json(same) = get_flows(State(state)).await.expect("get");
        assert_eq!(same.disabled_builtins, after.disabled_builtins);
    }
}