//! `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`; `0` when
    /// none is stored.
    revision: 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>,
    /// 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,
}

/// 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) -> Self {
        Self {
            revision,
            disabled_builtins: doc.disabled_builtins,
            builtins: notify_flows::builtins(),
            flows: doc.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),
        ))),
        Stored::Document(doc) => Ok(Json(FlowsView::of(doc, revision.unwrap_or(0)))),
        Stored::Unreadable(e) => Err(unreadable(&e)),
    }
}

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)));
        }
    }
    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(), 4);
        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(), 4, "3 built-ins 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":2,"flows":[]}"#.to_vec().into(),
            )
            .await
            .expect("seed");
        let got = get_flows(State(state.clone()))
            .await
            .expect_err("version 2");
        assert_eq!(got.status, Some(StatusCode::CONFLICT), "{got:?}");
        let refused = put_flows(State(state.clone()), Ok(body(0, &[], vec![])))
            .await
            .expect_err("version 2 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":2,"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");
    }
}