//! The router's two answers to swarm-controller: the newest deliveries and a
//! dry run (`swarm_wire::router_api`).
//!
//! Each is a NATS request. A dry run replays the newest forge events from the
//! events stream through [`dry_run::run`] and delivers nothing, writes no
//! snapshot and sends no review request. The router hears these on their own
//! task, so a slow replay never holds up the events.

use anyhow::Result;
use async_nats::jetstream;
use futures_util::StreamExt as _;
use serde::Serialize;
use swarm_wire::forge_events;
use swarm_wire::router_api::{
    DEFAULT_ROWS, DELIVERIES_SUBJECT, DRY_RUN_EVENTS, DRY_RUN_SUBJECT, Deliveries,
    DeliveriesRequest, DryRunRequest, Refused,
};

use crate::{dry_run, log};

/// Answer requests on both subjects until the connection closes.
pub async fn serve(client: async_nats::Client) {
    let js = jetstream::new(client.clone());
    let (mut deliveries, mut dry) = match (
        client.subscribe(DELIVERIES_SUBJECT).await,
        client.subscribe(DRY_RUN_SUBJECT).await,
    ) {
        (Ok(a), Ok(b)) => (a, b),
        (a, b) => {
            let why = a
                .err()
                .map(|e| e.to_string())
                .or(b.err().map(|e| e.to_string()));
            tracing::error!(error = ?why, "notify router: cannot hear api requests; the controller will report them unavailable");
            return;
        }
    };
    tracing::info!("notify router: answering deliveries and dry-run requests");
    loop {
        let (request, dry_run) = tokio::select! {
            Some(m) = deliveries.next() => (m, false),
            Some(m) = dry.next() => (m, true),
            else => return,
        };
        let (client, js) = (client.clone(), js.clone());
        tokio::spawn(async move {
            let Some(reply) = request.reply else { return };
            let answer = if dry_run {
                dry_run_answer(&js, &request.payload).await
            } else {
                deliveries_answer(&js, &request.payload).await
            };
            if let Err(e) = client.publish(reply, answer.into()).await {
                tracing::warn!(error = %e, "notify router: could not answer an api request");
            }
        });
    }
}

/// The reply body: `value`, or a [`Refused`] saying why not.
fn body<T: Serialize>(answer: Result<T>) -> Vec<u8> {
    match answer {
        Ok(value) => serde_json::to_vec(&value).expect("an answer encodes"),
        Err(e) => {
            tracing::warn!(
                error = format!("{e:#}"),
                "notify router: an api request failed"
            );
            serde_json::to_vec(&Refused {
                error: format!("{e:#}"),
            })
            .expect("a refusal encodes")
        }
    }
}

async fn deliveries_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
    body(
        async {
            let request: DeliveriesRequest = if payload.is_empty() {
                DeliveriesRequest::default()
            } else {
                serde_json::from_slice(payload)?
            };
            let limit = request.limit.unwrap_or(DEFAULT_ROWS);
            Ok(Deliveries {
                rows: log::recent(js, limit).await?,
            })
        }
        .await,
    )
}

async fn dry_run_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
    body(
        async {
            let request: DryRunRequest = serde_json::from_slice(payload)?;
            request
                .document
                .check()
                .map_err(|why| anyhow::anyhow!("the document fails its check: {why}"))?;
            let (flows, _) = request.document.active();
            let mut events: Vec<(String, Vec<u8>)> =
                log::newest(js, forge_events::STREAM, DRY_RUN_EVENTS)
                    .await?
                    .into_iter()
                    .filter_map(|(subject, body)| {
                        Some((forge_events::event_type(&subject)?.to_owned(), body))
                    })
                    .collect();
            events.reverse();
            Ok(dry_run::run(&flows, &events))
        }
        .await,
    )
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use swarm_wire::notify_flows::Document;
    use swarm_wire::router_api::{DryRun, Kind, LOG_STREAM, Row};
    use swarm_wire::router_todo;

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

    /// Send a request to the router's api, retrying while its subscription
    /// is not yet in place.
    async fn ask(client: &async_nats::Client, subject: &str, body: Vec<u8>) -> Vec<u8> {
        for _ in 0..50 {
            if let Ok(m) = client
                .request(subject.to_owned(), body.clone().into())
                .await
            {
                return m.payload.to_vec();
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        panic!("nothing answered {subject}");
    }

    fn row(n: u64) -> Row {
        log::Cause {
            event: "issue_assign".to_owned(),
            repo: "o/r".to_owned(),
            number: n,
        }
        .row(
            "assigned",
            "sink",
            Kind::Todo,
            "atlas",
            Ok("upsert k".to_owned()),
        )
    }

    async fn publish_event(js: &jetstream::Context, event: &str, n: i64) {
        let body = serde_json::json!({
            "action": "assigned",
            "repository": { "full_name": "o/r", "id": 1 },
            "issue": { "number": n, "title": "t", "html_url": "u", "assignees": [{ "login": "atlas" }] },
            "sender": { "login": "mara" },
        });
        js.publish(
            forge_events::subject(event),
            serde_json::to_vec(&body).unwrap().into(),
        )
        .await
        .expect("published")
        .await
        .expect("acked");
    }

    /// The newest rows come back newest first, no more than asked for, and
    /// the stream keeps 500 of them.
    #[tokio::test]
    async fn the_deliveries_are_the_newest_rows_and_the_log_keeps_500() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let js = jetstream::new(client.clone());
        log::open_or_create(&js).await.expect("the log stream");
        tokio::spawn(serve(client.clone()));
        for n in 1..=3 {
            log::append(&js, &row(n)).await;
        }

        let all: Deliveries =
            serde_json::from_slice(&ask(&client, DELIVERIES_SUBJECT, Vec::new()).await)
                .expect("an answer");
        let numbers: Vec<u64> = all.rows.iter().map(|r| r.number).collect();
        assert_eq!(numbers, [3, 2, 1]);
        let first = Row {
            at: all.rows[0].at,
            ..row(3)
        };
        assert_eq!(all.rows[0], first);

        let two: Deliveries = serde_json::from_slice(
            &ask(&client, DELIVERIES_SUBJECT, br#"{"limit":2}"#.to_vec()).await,
        )
        .expect("an answer");
        assert_eq!(
            two.rows.iter().map(|r| r.number).collect::<Vec<_>>(),
            [3, 2]
        );

        for n in 4..=503 {
            log::append(&js, &row(n)).await;
        }
        let mut stream = js.get_stream(LOG_STREAM).await.expect("the stream");
        assert_eq!(stream.info().await.expect("info").state.messages, 500);
        let newest: Deliveries = serde_json::from_slice(
            &ask(&client, DELIVERIES_SUBJECT, br#"{"limit":9999}"#.to_vec()).await,
        )
        .expect("an answer");
        assert_eq!(newest.rows.len(), 500);
        assert_eq!(newest.rows[0].number, 503);
        assert_eq!(newest.rows[499].number, 4);
    }

    /// A dry run counts the newest 50 events, and delivers nothing: no todo
    /// op, no log row.
    #[tokio::test]
    async fn a_dry_run_counts_the_newest_fifty_events_and_delivers_nothing() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let js = jetstream::new(client.clone());
        swarm_queue_client::forge_events::open_or_create(&client)
            .await
            .expect("events");
        swarm_queue_client::router_todo::open_or_create(&client)
            .await
            .expect("todos");
        log::open_or_create(&js).await.expect("the log stream");
        tokio::spawn(serve(client.clone()));
        // 60 assign events on distinct issues, then one on a type no flow
        // starts from.
        for n in 1..=60 {
            publish_event(&js, "issue_assign", n).await;
        }
        publish_event(&js, "issue_comment", 61).await;

        let request = serde_json::to_vec(&DryRunRequest {
            document: Document::default(),
        })
        .unwrap();
        let reply = ask(&client, DRY_RUN_SUBJECT, request).await;
        let done: DryRun = serde_json::from_slice(&reply).expect("a dry run");
        assert_eq!(done.events, 50);
        let hits = |flow: &str, node: &str| {
            done.flows
                .iter()
                .find(|f| f.flow == flow)
                .and_then(|f| f.nodes.iter().find(|n| n.node == node))
                .map(|n| n.hits)
        };
        // 49 of the newest 50 are assign events; the 50th is the comment.
        assert_eq!(hits("assigned", "source"), Some(49));
        assert_eq!(hits("assigned", "sink"), Some(49));
        assert_eq!(hits("unassigned", "sink"), Some(0));

        for (stream, what) in [(router_todo::STREAM, "todo op"), (LOG_STREAM, "log row")] {
            let mut s = js.get_stream(stream).await.expect("the stream");
            assert_eq!(s.info().await.expect("info").state.messages, 0, "a {what}");
        }
    }

    /// A document that fails its check is refused, and says why.
    #[tokio::test]
    async fn a_document_that_fails_its_check_is_refused() {
        let nats = TestNats::start();
        let client = nats.client().await;
        tokio::spawn(serve(client.clone()));
        let doc = Document {
            disabled_builtins: vec!["no-such-flow".to_owned()],
            ..Document::default()
        };
        let request = serde_json::to_vec(&DryRunRequest { document: doc }).unwrap();
        let refused: Refused =
            serde_json::from_slice(&ask(&client, DRY_RUN_SUBJECT, request).await)
                .expect("a refusal");
        assert!(refused.error.contains("no-such-flow"), "{}", refused.error);
        let garbled = ask(&client, DRY_RUN_SUBJECT, b"nonsense".to_vec()).await;
        assert!(serde_json::from_slice::<Refused>(&garbled).is_ok());
    }
}