//! The flows the router runs: the built-ins, changed by the document in the
//! `notify-flows` bucket.
//!
//! The router reads the key again every [`REFRESH`] and runs what
//! [`Document::active`] says. What it does with each thing it can read:
//!
//! | the key                              | the router runs                     |
//! | ------------------------------------ | ----------------------------------- |
//! | absent, deleted, or no such bucket   | the built-ins                       |
//! | a document it reads                  | its built-ins and custom flows      |
//! | a document it cannot read            | what it ran before                  |
//! | the read itself failing              | what it ran before                  |
//!
//! "Cannot read" is not JSON, not the document's shape, or a version this
//! build does not know. Before any document has been read, "what it ran
//! before" is the built-ins, so a bad document never leaves the router with
//! no flows.

use std::time::{Duration, Instant};

use async_nats::jetstream::ErrorCode;
use async_nats::jetstream::context::GetStreamErrorKind;
use async_nats::jetstream::{self, kv};
use swarm_wire::notify_flows::{self, Document, Flow};

/// How long a read of the key is reused.
const REFRESH: Duration = Duration::from_secs(10);

/// What a read of the key came to.
#[derive(Debug)]
pub enum Read {
    /// No key, a deleted key, or no bucket.
    Absent,
    Document {
        revision: u64,
        bytes: Vec<u8>,
    },
    Failed(String),
}

/// The flows in force, and which revision of the key they came from.
pub struct Flows {
    pub active: Vec<Flow>,
    /// The revision `active` was read from; `None` for the built-ins alone.
    revision: Option<u64>,
    /// The last problem logged, so a standing one is logged once.
    problem: Option<String>,
    checked: Option<Instant>,
}

impl Flows {
    /// The built-ins, which is what the router runs until it reads a document.
    pub fn builtins() -> Self {
        Self {
            active: Document::default().active().0,
            revision: None,
            problem: None,
            checked: None,
        }
    }

    /// Take what a read came to.
    pub fn apply(&mut self, read: Read) {
        match read {
            Read::Failed(why) => self.note(format!("reading the key failed: {why}")),
            Read::Absent => {
                if self.revision.take().is_some() {
                    tracing::info!(
                        "notify router: the flows document is gone; running the built-in flows"
                    );
                }
                self.active = Document::default().active().0;
                self.problem = None;
            }
            Read::Document { revision, .. } if self.revision == Some(revision) => {}
            Read::Document { revision, bytes } => match Document::decode(&bytes) {
                Err(why) => {
                    // The revision is not taken, so the next read looks at
                    // the document again and the problem is logged once.
                    self.note(format!("revision {revision} cannot be read: {why}"));
                }
                Ok(doc) => {
                    let (active, skipped) = doc.active();
                    for why in &skipped {
                        tracing::error!(%why, "notify router: skipping a flow");
                    }
                    tracing::info!(
                        revision,
                        flows = active.len(),
                        disabled = doc.disabled_builtins.len(),
                        custom = doc.flows.len(),
                        "notify router: running the flows document"
                    );
                    self.active = active;
                    self.revision = Some(revision);
                    self.problem = None;
                }
            },
        }
    }

    fn note(&mut self, problem: String) {
        if self.problem.as_ref() != Some(&problem) {
            tracing::error!(%problem, "notify router: keeping the flows it has");
            self.problem = Some(problem);
        }
    }

    /// Read the key if the last read is older than [`REFRESH`].
    pub async fn refresh(&mut self, js: &jetstream::Context) {
        if self.checked.is_some_and(|at| at.elapsed() < REFRESH) {
            return;
        }
        self.checked = Some(Instant::now());
        self.apply(read(js).await);
    }
}

async fn read(js: &jetstream::Context) -> Read {
    // A bucket that does not exist is the answer "no document". Any other
    // failure to look, such as a timeout, says nothing about the document.
    match js.get_stream(format!("KV_{}", notify_flows::BUCKET)).await {
        Ok(_) => {}
        Err(e)
            if matches!(e.kind(), GetStreamErrorKind::JetStream(j)
                if j.error_code() == ErrorCode::STREAM_NOT_FOUND) =>
        {
            return Read::Absent;
        }
        Err(e) => return Read::Failed(e.to_string()),
    }
    let store = match js.get_key_value(notify_flows::BUCKET).await {
        Ok(store) => store,
        Err(e) => return Read::Failed(e.to_string()),
    };
    match store.entry(notify_flows::KEY).await {
        Ok(Some(entry)) if entry.operation == kv::Operation::Put => Read::Document {
            revision: entry.revision,
            bytes: entry.value.to_vec(),
        },
        Ok(_) => Read::Absent,
        Err(e) => Read::Failed(e.to_string()),
    }
}

#[cfg(test)]
mod tests {
    use swarm_wire::notify_flows::{Document, VERSION};

    use super::*;

    fn ids(f: &Flows) -> Vec<&str> {
        f.active.iter().map(|f| f.id.as_str()).collect()
    }

    fn doc(disabled: &[&str], custom: &[&str]) -> Read {
        let mut flows = Vec::new();
        for id in custom {
            let mut f = notify_flows::builtins().remove(0);
            f.id = (*id).to_owned();
            flows.push(f);
        }
        let d = Document {
            version: VERSION,
            disabled_builtins: disabled.iter().map(|s| (*s).to_owned()).collect(),
            flows,
        };
        Read::Document {
            revision: 0,
            bytes: serde_json::to_vec(&d).unwrap(),
        }
    }

    fn at(revision: u64, read: Read) -> Read {
        match read {
            Read::Document { bytes, .. } => Read::Document { revision, bytes },
            other => other,
        }
    }

    fn junk(revision: u64, bytes: &[u8]) -> Read {
        Read::Document {
            revision,
            bytes: bytes.to_vec(),
        }
    }

    #[test]
    fn with_no_document_the_router_runs_the_builtins() {
        let mut f = Flows::builtins();
        assert_eq!(
            ids(&f),
            ["assigned", "unassigned", "closed", "head-classifier"]
        );
        f.apply(Read::Absent);
        assert_eq!(
            ids(&f),
            ["assigned", "unassigned", "closed", "head-classifier"]
        );
        // An empty document is the same.
        f.apply(at(1, doc(&[], &[])));
        assert_eq!(
            ids(&f),
            ["assigned", "unassigned", "closed", "head-classifier"]
        );
    }

    #[test]
    fn a_document_switches_builtins_off_and_adds_custom_flows() {
        let mut f = Flows::builtins();
        f.apply(at(1, doc(&["unassigned"], &["triage"])));
        assert_eq!(ids(&f), ["assigned", "closed", "head-classifier", "triage"]);
    }

    #[test]
    fn a_document_it_cannot_read_leaves_the_flows_as_they_were() {
        let mut f = Flows::builtins();
        f.apply(at(1, doc(&["closed"], &["triage"])));
        let before = ids(&f).join(",");
        assert_eq!(before, "assigned,unassigned,head-classifier,triage");
        for bad in [
            junk(2, b"not json"),
            junk(3, br#"{"version":2,"flows":[]}"#),
            junk(4, br#"{"flows":[{"id":"x"}]}"#),
            Read::Failed("timeout".to_owned()),
        ] {
            f.apply(bad);
            assert_eq!(ids(&f).join(","), before);
        }
        // The next good revision replaces them: the control that the bad
        // ones were kept out by being bad, not by the loader being stuck.
        f.apply(at(5, doc(&[], &[])));
        assert_eq!(
            ids(&f),
            ["assigned", "unassigned", "closed", "head-classifier"]
        );
    }

    #[test]
    fn a_bad_first_document_is_the_builtins_not_nothing() {
        let mut f = Flows::builtins();
        f.apply(junk(1, br#"{"version":9}"#));
        assert_eq!(
            ids(&f),
            ["assigned", "unassigned", "closed", "head-classifier"]
        );
    }

    #[test]
    fn a_deleted_document_goes_back_to_the_builtins() {
        let mut f = Flows::builtins();
        f.apply(at(1, doc(&["closed"], &["triage"])));
        f.apply(Read::Absent);
        assert_eq!(
            ids(&f),
            ["assigned", "unassigned", "closed", "head-classifier"]
        );
    }

    #[test]
    fn a_flow_that_is_not_valid_is_left_out_and_the_rest_run() {
        let mut f = Flows::builtins();
        let mut broken = notify_flows::builtins().remove(0);
        broken.id = "broken".to_owned();
        broken.edges.clear();
        let d = Document {
            version: VERSION,
            disabled_builtins: Vec::new(),
            flows: vec![broken],
        };
        f.apply(Read::Document {
            revision: 1,
            bytes: serde_json::to_vec(&d).unwrap(),
        });
        assert_eq!(
            ids(&f),
            ["assigned", "unassigned", "closed", "head-classifier"]
        );
    }
}