//! The deliveries log: one row on the `NOTIFY_ROUTER_LOG` stream for each
//! delivery the router makes (`swarm_wire::router_api`), and reading the
//! newest rows back.
//!
//! The stream keeps the newest 500 rows. Appending is best effort: a log that
//! cannot be written is reported and never holds up a delivery.

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::{Context as _, Result};
use async_nats::jetstream::stream::{Config, DiscardPolicy, Stream};
use async_nats::jetstream::{self};
use futures_util::{StreamExt as _, stream};
use swarm_wire::router_api::{Kind, LOG_MAX_MSGS, LOG_STREAM, LOG_SUBJECT, Row};

/// How long an append may take before it is given up on.
const APPEND_WITHIN: Duration = Duration::from_secs(2);

/// How many messages are fetched at once.
const FETCH_AT_ONCE: usize = 16;

/// Open the stream, creating it if nothing has yet. The router is its one
/// writer and its one creator.
pub async fn open_or_create(js: &jetstream::Context) -> Result<Stream> {
    match js.get_stream(LOG_STREAM).await {
        Ok(stream) => Ok(stream),
        Err(e) => {
            tracing::info!(stream = LOG_STREAM, reason = %e, "notify router: log stream not available, creating it");
            js.create_stream(Config {
                name: LOG_STREAM.to_owned(),
                description: Some("Notification router deliveries, newest 500".to_owned()),
                subjects: vec![LOG_SUBJECT.to_owned()],
                max_messages: LOG_MAX_MSGS,
                discard: DiscardPolicy::Old,
                ..Default::default()
            })
            .await
            .context("creating the log stream")
        }
    }
}

/// The event a delivery answers.
#[derive(Clone, Debug)]
pub struct Cause {
    pub event: String,
    pub repo: String,
    pub number: u64,
}

impl Cause {
    /// A row of `kind` made at `flow`'s `node` for `target`, which went
    /// through when `result` is `Ok`.
    pub fn row(
        &self,
        flow: &str,
        node: &str,
        kind: Kind,
        target: &str,
        result: Result<String, String>,
    ) -> Row {
        let (ok, detail) = match result {
            Ok(detail) => (true, detail),
            Err(why) => (false, why),
        };
        Row {
            at: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map_or(0, |d| d.as_secs()),
            flow: flow.to_owned(),
            node: node.to_owned(),
            event: self.event.clone(),
            repo: self.repo.clone(),
            number: self.number,
            kind,
            target: target.to_owned(),
            ok,
            detail,
        }
    }
}

/// Log one routed delivery: at info when `flows` flows produced a todo or a
/// review request, at debug when none did, so deliveries nothing wants do not
/// flood the log. It names the sender and never the body.
pub fn delivery(
    event_type: &str,
    delivery: &crate::event::Delivery,
    delivery_id: Option<&str>,
    flows: usize,
) {
    let (repo, number, sender, action) = (
        &delivery.repo,
        delivery.number,
        &delivery.sender,
        &delivery.action,
    );
    let delivery_id = delivery_id.unwrap_or("none");
    if flows == 0 {
        tracing::debug!(event_type, %action, %repo, number, %sender, delivery_id, flows, "notify router: forge event matched no flow");
    } else {
        tracing::info!(event_type, %action, %repo, number, %sender, delivery_id, flows, "notify router: forge event routed");
    }
}

/// Log what one sink of `flow` did for `target` on the issue or PR `cause` names: info when it went through,
/// warn with the error when it did not.
pub fn sink(cause: &Cause, flow: &str, kind: Kind, target: &str, result: &Result<String, String>) {
    let (repo, number) = (&cause.repo, cause.number);
    match result {
        Ok(outcome) => {
            tracing::info!(%flow, %repo, number, ?kind, %target, %outcome, "notify router: sink ran");
        }
        Err(error) => {
            tracing::warn!(%flow, %repo, number, ?kind, %target, %error, "notify router: sink failed");
        }
    }
}

/// Append `row`; a failure, or an append that takes longer than
/// `APPEND_WITHIN`, is logged and dropped. It is awaited in place, so rows
/// keep the order of the deliveries.
pub async fn append(js: &jetstream::Context, row: &Row) {
    append_within(js, row, APPEND_WITHIN).await;
}

async fn append_within(js: &jetstream::Context, row: &Row, within: Duration) {
    let body = serde_json::to_vec(row).expect("a row encodes");
    let sent = async {
        js.publish(LOG_SUBJECT, body.into())
            .await
            .context("publishing")?
            .await
            .context("waiting for the ack")
    };
    let failure = match tokio::time::timeout(within, sent).await {
        Ok(Ok(_)) => return,
        Ok(Err(e)) => format!("{e:#}"),
        Err(_) => format!("no ack within {within:?}"),
    };
    tracing::warn!(error = failure, flow = %row.flow, "notify router: could not log a delivery");
}

/// The payloads of the newest `limit` messages of `stream`, newest first,
/// with their subjects. A sequence that no longer exists is skipped.
pub async fn newest(
    js: &jetstream::Context,
    stream: &str,
    limit: usize,
) -> Result<Vec<(String, Vec<u8>)>> {
    let stream = js
        .get_stream(stream)
        .await
        .with_context(|| format!("opening {stream}"))?;
    let state = &stream.cached_info().state;
    let (first, last) = (state.first_sequence, state.last_sequence);
    let count = u64::try_from(limit).unwrap_or(u64::MAX);
    let from = first.max((last + 1).saturating_sub(count));
    let seqs = (from..=last).rev();
    let fetched: Vec<_> = stream::iter(seqs)
        .map(|seq| {
            let stream = &stream;
            async move { stream.get_raw_message(seq).await }
        })
        .buffered(FETCH_AT_ONCE)
        .collect()
        .await;
    Ok(fetched
        .into_iter()
        .filter_map(Result::ok)
        .map(|m| (m.subject.to_string(), m.payload.to_vec()))
        .collect())
}

/// The newest `limit` rows of the log, newest first. A row that does not
/// decode is skipped.
pub async fn recent(js: &jetstream::Context, limit: usize) -> Result<Vec<Row>> {
    Ok(newest(js, LOG_STREAM, limit)
        .await?
        .into_iter()
        .filter_map(|(_, body)| serde_json::from_slice(&body).ok())
        .collect())
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};
    use std::time::Instant;

    use tracing_subscriber::layer::SubscriberExt as _;

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

    /// The level and message of every event.
    #[derive(Clone, Default)]
    struct Messages(Arc<Mutex<Vec<(tracing::Level, String)>>>);

    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Messages {
        fn on_event(
            &self,
            event: &tracing::Event<'_>,
            _: tracing_subscriber::layer::Context<'_, S>,
        ) {
            let mut message = String::new();
            event.record(
                &mut |field: &tracing::field::Field, value: &dyn std::fmt::Debug| {
                    if field.name() == "message" {
                        message = format!("{value:?}");
                    }
                },
            );
            self.0
                .lock()
                .expect("unpoisoned")
                .push((*event.metadata().level(), message));
        }
    }

    /// The messages logged at warn or above.
    fn warnings(messages: &Messages) -> Vec<String> {
        messages
            .0
            .lock()
            .expect("unpoisoned")
            .iter()
            .filter(|(level, _)| *level <= tracing::Level::WARN)
            .map(|(_, m)| m.clone())
            .collect()
    }

    fn a_row() -> Row {
        Cause {
            event: "issue_assign".to_owned(),
            repo: "o/r".to_owned(),
            number: 1,
        }
        .row("f", "n", Kind::Todo, "agent", Ok("upsert k".to_owned()))
    }

    /// A subscriber on the log subject with no stream behind it: the publish
    /// goes out and no ack ever comes back.
    #[tokio::test]
    async fn an_append_nothing_acks_gives_up_and_warns() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let js = jetstream::new(client.clone());
        let _silent = client.subscribe(LOG_SUBJECT).await.expect("subscribed");
        let messages = Messages::default();
        let _guard =
            tracing::subscriber::set_default(tracing_subscriber::registry().with(messages.clone()));

        let started = Instant::now();
        append_within(&js, &a_row(), Duration::from_millis(300)).await;
        assert!(
            started.elapsed() < Duration::from_secs(2),
            "{:?}",
            started.elapsed()
        );
        assert_eq!(
            warnings(&messages),
            ["notify router: could not log a delivery"]
        );
    }

    /// Control: with the stream in place the same call acks and logs nothing.
    #[tokio::test]
    async fn an_append_the_stream_acks_logs_nothing() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let js = jetstream::new(client);
        open_or_create(&js).await.expect("the log stream");
        let messages = Messages::default();
        let _guard =
            tracing::subscriber::set_default(tracing_subscriber::registry().with(messages.clone()));

        append_within(&js, &a_row(), Duration::from_secs(5)).await;
        assert!(warnings(&messages).is_empty());
        assert_eq!(recent(&js, 10).await.expect("rows").len(), 1);
    }
}