//! 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. A `request_reviewer` node is a leaf that asks the forge for
//! a review of the pull request the event concerns. 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, with what each means:
/// the ones swarm-controller's hooks deliver to the events stream.
pub const SOURCE_EVENTS: &[(&str, &str)] = &[
    (
        "issues",
        "An issue opened, closed, edited or otherwise changed",
    ),
    ("issue_assign", "An issue assigned or unassigned"),
    (
        "pull_request",
        "A pull request opened, closed, edited or otherwise changed",
    ),
    (
        "pull_request_assign",
        "A pull request assigned or unassigned",
    ),
    (
        "pull_request_sync",
        "A push to the branch of an open pull request",
    ),
    (
        "pull_request_review_approved",
        "A review submitted on a pull request that approves it",
    ),
    (
        "pull_request_review_rejected",
        "A review submitted on a pull request that asks for changes",
    ),
    (
        "pull_request_review_comment",
        "A review submitted on a pull request that only comments",
    ),
];

/// The fields of an event a condition or a recipient can name, with what each
/// holds. A todo's key and text read them as `{field}` placeholders.
pub const FIELDS: &[(&str, &str)] = &[
    ("event", "The event type, one of the source events"),
    ("action", "What happened, such as opened, closed, assigned"),
    ("kind", "issue or pull request"),
    ("repo", "The repository, owner/name"),
    ("number", "The issue or pull request number"),
    ("title", "The issue or pull request title"),
    ("url", "The web link to the issue or pull request"),
    ("sender", "The account that caused the event"),
    (
        "draft",
        "true when the pull request is a draft (its title starts WIP: or [WIP]), else false",
    ),
    (
        "undrafted",
        "true when the event is a title edit that took the pull request out of draft, else false",
    ),
    ("assignees", "The agents assigned after the event (a list)"),
    ("assignees_added", "The agents the event assigned (a list)"),
    (
        "assignees_removed",
        "The agents the event unassigned (a list)",
    ),
    (
        "todo_holders",
        "The agents holding a router todo for the issue or pull request (a list)",
    ),
];

/// The forge account of the review classifier, whose review the built-in
/// `head-classifier` flow requests.
pub const CLASSIFIER_ACCOUNT: &str = "review-classifier";

/// The todo ops a `todo` node's `op` takes, one per [`Op`] variant.
const OPS: &[(&str, &str)] = &[
    (
        "upsert",
        "Add the todo, or update the one under the same key",
    ),
    (
        "clear",
        "Remove the todo under the key, or with prefix every one starting with it",
    ),
];

/// 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<String>,
    /// The flows the operator defined.
    #[serde(default)]
    pub flows: Vec<Flow>,
}

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:<id>`.
    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<Node>,
    pub edges: Vec<Edge>,
}

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<String> },
    /// Passes when every condition holds.
    Match { when: Vec<Condition> },
    /// Emits a todo op to each recipient.
    Todo(TodoSink),
    /// Asks the forge for `reviewer`'s review of the pull request the event
    /// concerns, unless their review of its head already stands. Does nothing
    /// for an issue.
    RequestReviewer {
        /// A forge login.
        reviewer: String,
    },
}

/// `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<Self, DecodeError> {
        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<Flow>, Vec<String>) {
        let mut flows: Vec<Flow> = 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'_')
}

/// Whether `login` can name a forge account: Forgejo allows letters, digits,
/// `-`, `_` and `.`, and no more than 40 of them.
fn is_login(login: &str) -> bool {
    !login.is_empty()
        && login.len() <= 40
        && login
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.'))
}

fn is_field(name: &str) -> bool {
    FIELDS.iter().any(|(n, _)| *n == name)
}

fn is_source_event(name: &str) -> bool {
    SOURCE_EVENTS.iter().any(|(n, _)| *n == name)
}

/// A name with a one-line description.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct Described {
    pub name: String,
    pub description: String,
}

/// `values_from` marker: the names of the agents, as `GET /api/agents` lists
/// them.
pub const VALUES_FROM_AGENTS: &str = "agents";

/// `values_from` marker: the repositories with an open issue, as
/// `GET /api/repos` lists them. A repository with none is missing, so a field
/// that takes this marker is never closed.
pub const VALUES_FROM_REPOS: &str = "repos";

/// The `kind` field's values: what the router puts there, which is not the
/// words of the field's description.
const KINDS: &[(&str, &str)] = &[("issue", "An issue"), ("PR", "A pull request")];

/// Some of what the forge sends as `action`. Others exist, so it is a list of
/// suggestions only.
const ACTIONS: &[(&str, &str)] = &[
    ("opened", "Created"),
    ("closed", "Closed, or a pull request merged"),
    ("reopened", "Opened again after being closed"),
    ("edited", "Title or text changed"),
    ("assigned", "An assignee added"),
    ("unassigned", "An assignee removed"),
    ("synchronized", "New commits pushed to a pull request"),
];

/// One event field a condition or recipient can name, and what it can hold.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct EventField {
    pub name: String,
    pub description: String,
    /// Values the field is known to hold; empty when none are listed.
    pub values: Vec<Described>,
    /// Where an editor asks for the values only known live: one of
    /// [`VALUES_FROM_AGENTS`], [`VALUES_FROM_REPOS`]. Absent when there is
    /// none.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub values_from: Option<String>,
    /// Whether `values` or the source is every value the field can hold,
    /// which an editor can show as a dropdown. When false they are
    /// suggestions and the editor still takes any other text.
    pub closed: bool,
}

/// One field of a node kind.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct KindField {
    pub name: String,
    pub description: String,
    /// Whether a node of the kind must set it.
    pub required: bool,
    /// The accepted values when the field takes one of a fixed few; empty
    /// otherwise.
    pub values: Vec<Described>,
    /// Where an editor asks for the values only known live, as
    /// [`EventField::values_from`].
    #[serde(skip_serializing_if = "Option::is_none")]
    pub values_from: Option<String>,
    /// Whether the values are every one the field accepts, as
    /// [`EventField::closed`].
    pub closed: bool,
}

/// One kind of node and the fields it takes besides `id` and `kind`.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct KindSchema {
    pub kind: String,
    pub description: String,
    pub fields: Vec<KindField>,
}

/// The limits [`Document::check`] enforces.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct Limits {
    pub max_nodes: usize,
    pub max_depth: usize,
    pub max_flows: usize,
}

/// What a flow editor needs to offer only what [`Document::check`] accepts,
/// built from the constants the check reads.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
pub struct Schema {
    pub source_events: Vec<Described>,
    pub fields: Vec<EventField>,
    pub node_kinds: Vec<KindSchema>,
    pub limits: Limits,
}

fn described(list: &[(&str, &str)]) -> Vec<Described> {
    list.iter()
        .map(|(name, description)| Described {
            name: (*name).to_owned(),
            description: (*description).to_owned(),
        })
        .collect()
}

fn kind_field(name: &str, description: &str, required: bool) -> KindField {
    KindField {
        name: name.to_owned(),
        description: description.to_owned(),
        required,
        values: Vec::new(),
        values_from: None,
        closed: false,
    }
}

/// [`FIELDS`] with what each is known to hold.
fn event_fields() -> Vec<EventField> {
    FIELDS
        .iter()
        .map(|(name, description)| {
            let (values, values_from, closed) = match *name {
                "event" => (described(SOURCE_EVENTS), None, true),
                "kind" => (described(KINDS), None, true),
                "action" => (described(ACTIONS), None, false),
                "repo" => (Vec::new(), Some(VALUES_FROM_REPOS), false),
                "assignees" | "assignees_added" | "assignees_removed" | "todo_holders" => {
                    (Vec::new(), Some(VALUES_FROM_AGENTS), false)
                }
                _ => (Vec::new(), None, false),
            };
            EventField {
                name: (*name).to_owned(),
                description: (*description).to_owned(),
                values,
                values_from: values_from.map(str::to_owned),
                closed,
            }
        })
        .collect()
}

/// The schema of this build's flows.
#[must_use]
pub fn schema() -> Schema {
    let mut op = kind_field("op", "What the todo does to the recipient's list", true);
    op.values = described(OPS);
    op.closed = true;
    let mut to = kind_field(
        "to",
        "The recipients: {field: <event field>} or {agent: <agent name>}",
        true,
    );
    to.values_from = Some(VALUES_FROM_AGENTS.to_owned());
    to.closed = true;
    let mut reviewer = kind_field("reviewer", "The forge login to request a review from", true);
    // A login need not belong to an agent: the classifier's does not.
    reviewer.values_from = Some(VALUES_FROM_AGENTS.to_owned());
    Schema {
        source_events: described(SOURCE_EVENTS),
        fields: event_fields(),
        node_kinds: vec![
            KindSchema {
                kind: "source".to_owned(),
                description: "The root: the forge events the flow starts from".to_owned(),
                fields: vec![kind_field(
                    "events",
                    "One or more of the source events",
                    true,
                )],
            },
            KindSchema {
                kind: "match".to_owned(),
                description: "Passes when every condition holds; each child then runs".to_owned(),
                fields: vec![kind_field(
                    "when",
                    "Conditions, each {field, equals}; a list field holds when one item equals",
                    true,
                )],
            },
            KindSchema {
                kind: "todo".to_owned(),
                description: "A leaf: emits one todo op per recipient".to_owned(),
                fields: vec![
                    to,
                    op,
                    kind_field("key", "The todo key, with {field} placeholders", true),
                    kind_field("prefix", "For a clear: key is a prefix", false),
                    kind_field(
                        "text",
                        "The todo text, with {field} placeholders; required for an upsert, empty for a clear",
                        false,
                    ),
                ],
            },
            KindSchema {
                kind: "request_reviewer".to_owned(),
                description: "A leaf: asks the forge for a review of the pull request the event concerns, unless the reviewer's review of its head already stands; does nothing for an issue".to_owned(),
                fields: vec![reviewer],
            },
        ],
        limits: Limits {
            max_nodes: MAX_NODES,
            max_depth: MAX_DEPTH,
            max_flows: MAX_FLOWS,
        },
    }
}

/// The `{name}` placeholders in `template`, in order. A `{` with no `}` after
/// it is plain text.
#[must_use]
pub fn placeholders(template: &str) -> Vec<&str> {
    let mut names = Vec::new();
    let mut rest = template;
    while let Some(open) = rest.find('{') {
        let after = &rest[open + 1..];
        let Some(close) = after.find('}') else { break };
        names.push(&after[..close]);
        rest = &after[close + 1..];
    }
    names
}

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(_) | NodeKind::RequestReviewer { .. }
            ) && !self.children(&node.id).is_empty()
            {
                return Err(format!("{} has children, and is a leaf", 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| !is_source_event(e)) {
                    return Err(format!(
                        "event {bad:?} is not one of {}",
                        SOURCE_EVENTS
                            .iter()
                            .map(|(n, _)| *n)
                            .collect::<Vec<_>>()
                            .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()?,
            Self::RequestReviewer { reviewer } => {
                if !is_login(reviewer) {
                    return Err(format!("reviewer {reviewer:?} is not a forge login"));
                }
            }
        }
        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"));
        }
        for (what, template) in [("key", &self.key), ("text", &self.text)] {
            if let Some(bad) = placeholders(template).into_iter().find(|n| !is_field(n)) {
                return Err(format!("{what} has {{{bad}}}, which is no event field"));
            }
        }
        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,
///   then tells each agent that held one with a `closed` todo. The clear is the
///   first child of the match and runs first: its prefix also covers the
///   `closed` key.
/// - `head-classifier`: a PR opened, reopened or pushed to requests the review
///   classifier's review.
#[must_use]
pub fn builtins() -> Vec<Flow> {
    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(),
            },
        ),
        closed(),
        head_classifier(),
    ]
}

/// The `closed` flow: the clear, then the todo that says why.
fn closed() -> Flow {
    let mut flow = 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(),
        },
    );
    flow.nodes.push(Node {
        id: "inform".to_owned(),
        kind: NodeKind::Todo(TodoSink {
            to: Recipients::Field("todo_holders".to_owned()),
            op: Op::Upsert,
            key: "forge:{repo}#{number}:closed".to_owned(),
            prefix: false,
            text: "{kind} {repo}#{number} was closed: {title} ({url})".to_owned(),
        }),
    });
    flow.edges.push(Edge {
        from: "match".to_owned(),
        to: "inform".to_owned(),
    });
    flow
}

/// source(`pull_request`, `pull_request_sync`) → match(`action` = `opened`,
/// `reopened` or `synchronized`, and `draft` = `false`) → request the review
/// classifier's review; and match(`undrafted` = `true`) → the same request,
/// for a PR whose last push came while it was a draft.
fn head_classifier() -> Flow {
    let branches = [
        ("opened", "action", "opened"),
        ("reopened", "action", "reopened"),
        ("synchronized", "action", "synchronized"),
        ("undrafted", "undrafted", "true"),
    ];
    let mut nodes = vec![Node {
        id: "source".to_owned(),
        kind: NodeKind::Source {
            events: vec!["pull_request".to_owned(), "pull_request_sync".to_owned()],
        },
    }];
    let mut edges = Vec::new();
    for (name, field, equals) in branches {
        let (matched, request) = (format!("match-{name}"), format!("request-{name}"));
        let condition = |field: &str, equals: &str| Condition {
            field: field.to_owned(),
            equals: equals.to_owned(),
        };
        let mut when = vec![condition(field, equals)];
        if field == "action" {
            when.push(condition("draft", "false"));
        }
        nodes.push(Node {
            id: matched.clone(),
            kind: NodeKind::Match { when },
        });
        nodes.push(Node {
            id: request.clone(),
            kind: NodeKind::RequestReviewer {
                reviewer: CLASSIFIER_ACCOUNT.to_owned(),
            },
        });
        edges.push(Edge {
            from: "source".to_owned(),
            to: matched.clone(),
        });
        edges.push(Edge {
            from: matched,
            to: request,
        });
    }
    Flow {
        id: "head-classifier".to_owned(),
        name: "head-classifier".to_owned(),
        enabled: true,
        nodes,
        edges,
    }
}

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

    /// What the schema offers is what the check accepts, and nothing else is.
    #[test]
    fn the_schema_lists_what_a_flow_may_name() {
        let s = schema();
        let with_event = |name: &str| {
            let mut f = builtin("assigned");
            f.nodes[0].kind = NodeKind::Source {
                events: vec![name.to_owned()],
            };
            f.validate()
        };
        assert!(!s.source_events.is_empty());
        for e in &s.source_events {
            assert_eq!(with_event(&e.name), Ok(()), "{}", e.name);
            assert!(!e.description.is_empty(), "{}", e.name);
        }
        assert!(with_event("no_such_event").is_err(), "control");

        let with_field = |name: &str| {
            let mut f = builtin("assigned");
            f.nodes[1].kind = NodeKind::Match {
                when: vec![Condition {
                    field: name.to_owned(),
                    equals: "x".to_owned(),
                }],
            };
            f.validate()
        };
        assert!(!s.fields.is_empty());
        for field in &s.fields {
            assert_eq!(with_field(&field.name), Ok(()), "{}", field.name);
            assert!(!field.description.is_empty(), "{}", field.name);
        }
        assert!(with_field("no_such_field").is_err(), "control");

        let kinds: Vec<&str> = s.node_kinds.iter().map(|k| k.kind.as_str()).collect();
        assert_eq!(kinds, ["source", "match", "todo", "request_reviewer"]);
        let op_names = |op: &str| -> Option<Op> {
            serde_json::from_value(serde_json::Value::String(op.to_owned())).ok()
        };
        let todo = s
            .node_kinds
            .iter()
            .find(|k| k.kind == "todo")
            .expect("todo");
        let op = todo.fields.iter().find(|f| f.name == "op").expect("op");
        let listed: Vec<Op> = op
            .values
            .iter()
            .map(|v| op_names(&v.name).expect("an op"))
            .collect();
        // Every `Op` variant is listed: this match stops compiling when one is added.
        for o in [Op::Upsert, Op::Clear] {
            match o {
                Op::Upsert | Op::Clear => assert!(listed.contains(&o), "{o:?}"),
            }
        }
        assert_eq!(listed.len(), 2);
        assert!(op_names("nope").is_none(), "control");

        // What an editor offers for a field is what the router puts there.
        let field = |name: &str| s.fields.iter().find(|f| f.name == name).expect(name);
        let names =
            |f: &EventField| -> Vec<String> { f.values.iter().map(|v| v.name.clone()).collect() };
        assert_eq!(names(field("kind")), ["issue", "PR"]);
        assert!(field("kind").closed);
        let events: Vec<String> = s.source_events.iter().map(|e| e.name.clone()).collect();
        assert_eq!(names(field("event")), events);
        assert!(field("event").closed);
        assert!(!field("action").closed, "the forge may send more actions");
        assert!(names(field("action")).contains(&"closed".to_owned()));
        assert_eq!(
            field("repo").values_from.as_deref(),
            Some(VALUES_FROM_REPOS)
        );
        assert_eq!(
            field("assignees").values_from.as_deref(),
            Some(VALUES_FROM_AGENTS)
        );
        assert!(field("title").values.is_empty() && field("title").values_from.is_none());
        // A marker is one an editor knows to resolve.
        for f in &s.fields {
            if let Some(from) = &f.values_from {
                assert!([VALUES_FROM_AGENTS, VALUES_FROM_REPOS].contains(&from.as_str()));
            }
        }
        let kind_field_of = |kind: &str, name: &str| {
            s.node_kinds
                .iter()
                .find(|k| k.kind == kind)
                .and_then(|k| k.fields.iter().find(|f| f.name == name))
                .expect(name)
        };
        assert!(kind_field_of("todo", "op").closed);
        assert_eq!(
            kind_field_of("todo", "to").values_from.as_deref(),
            Some(VALUES_FROM_AGENTS)
        );
        let reviewer = kind_field_of("request_reviewer", "reviewer");
        assert_eq!(reviewer.values_from.as_deref(), Some(VALUES_FROM_AGENTS));
        assert!(!reviewer.closed, "any forge login is allowed");

        assert_eq!(
            (s.limits.max_nodes, s.limits.max_depth, s.limits.max_flows),
            (MAX_NODES, MAX_DEPTH, MAX_FLOWS)
        );
    }

    #[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_syn".to_owned()],
        };
        assert!(event.validate().unwrap_err().contains("pull_request_syn"));
        // Control: the spelling that exists passes.
        event.nodes[0].kind = NodeKind::Source {
            events: vec!["pull_request_sync".to_owned()],
        };
        assert_eq!(event.validate(), Ok(()));

        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(()));
    }

    /// A reviewer request is a leaf naming a forge login; the wire spells the
    /// kind as the schema lists it.
    #[test]
    fn a_request_reviewer_node_is_a_leaf_naming_a_login() {
        let with = |reviewer: &str| {
            let mut f = custom("c");
            f.nodes[2].kind = NodeKind::RequestReviewer {
                reviewer: reviewer.to_owned(),
            };
            f.validate()
        };
        assert_eq!(with("review-classifier"), Ok(()));
        assert_eq!(with("a.b_c-1"), Ok(()));
        for bad in ["", "two words", "a/b", "x@y", &"a".repeat(41)] {
            assert!(with(bad).is_err(), "{bad:?}");
        }

        let mut parent = custom("c");
        parent.nodes[2].kind = NodeKind::RequestReviewer {
            reviewer: "review-classifier".to_owned(),
        };
        parent.nodes.push(Node {
            id: "m2".to_owned(),
            kind: NodeKind::Match { when: Vec::new() },
        });
        parent.edges.push(edge("sink", "m2"));
        assert!(
            parent.validate().unwrap_err().contains("leaf"),
            "a reviewer request with a child"
        );

        let json = serde_json::to_value(Node {
            id: "ask".to_owned(),
            kind: NodeKind::RequestReviewer {
                reviewer: "review-classifier".to_owned(),
            },
        })
        .unwrap();
        assert_eq!(
            json,
            serde_json::json!({
                "id": "ask",
                "kind": "request_reviewer",
                "reviewer": "review-classifier",
            })
        );
        let kind = json["kind"].as_str().unwrap();
        assert!(
            schema().node_kinds.iter().any(|k| k.kind == kind),
            "the schema lists the kind the wire uses"
        );
        let node: Node = serde_json::from_value(json).unwrap();
        assert!(matches!(node.kind, NodeKind::RequestReviewer { .. }));
    }

    /// The built-in that requests the classifier on an open, a reopen and a push.
    #[test]
    fn head_classifier_requests_the_classifier_on_open_reopen_and_push() {
        let f = builtin("head-classifier");
        assert_eq!(f.validate(), Ok(()));
        let cond = |field: &str, equals: &str| Condition {
            field: field.to_owned(),
            equals: equals.to_owned(),
        };
        assert_eq!(
            f.nodes[0].kind,
            NodeKind::Source {
                events: vec!["pull_request".to_owned(), "pull_request_sync".to_owned()]
            }
        );
        assert_eq!(
            f.children("source"),
            [
                "match-opened",
                "match-reopened",
                "match-synchronized",
                "match-undrafted"
            ]
        );
        for action in ["opened", "reopened", "synchronized", "undrafted"] {
            let when = if action == "undrafted" {
                vec![cond("undrafted", "true")]
            } else {
                vec![cond("action", action), cond("draft", "false")]
            };
            assert_eq!(
                f.node(&format!("match-{action}")).map(|n| &n.kind),
                Some(&NodeKind::Match { when })
            );
            assert_eq!(
                f.node(&format!("request-{action}")).map(|n| &n.kind),
                Some(&NodeKind::RequestReviewer {
                    reviewer: CLASSIFIER_ACCOUNT.to_owned()
                })
            );
            assert_eq!(
                f.children(&format!("match-{action}")),
                [format!("request-{action}")]
            );
        }
    }

    #[test]
    fn a_placeholder_must_name_an_event_field() {
        let with = |key: &str, text: &str| {
            let mut f = custom("c");
            f.nodes[2].kind = NodeKind::Todo(TodoSink {
                to: Recipients::Agent("triage".to_owned()),
                op: Op::Upsert,
                key: key.to_owned(),
                prefix: false,
                text: text.to_owned(),
            });
            f.validate()
        };
        assert_eq!(
            with("forge:{repo}#{number}:triage", "{title} {url}"),
            Ok(())
        );
        assert!(
            with("forge:{repo}#{numbr}:triage", "t")
                .unwrap_err()
                .contains("numbr")
        );
        assert!(with("k", "see {ulr}").unwrap_err().contains("ulr"));
        assert!(with("k{}", "t").is_err(), "an empty name");
        // An unclosed brace is text.
        assert_eq!(with("k", "set {x"), Ok(()));
        assert_eq!(placeholders("a{b}c{d} {e"), ["b", "d"]);
    }

    #[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<Flow>| 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::<Vec<_>>();
        assert_eq!(
            ids(&Document::default()),
            ["assigned", "unassigned", "closed", "head-classifier"],
            "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", "head-classifier", "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": "",
            })
        );
    }
}