//! Handling one pulled batch in stream order.
//!
//! An issue's snapshot diff depends on the order its events are handled in,
//! so a failed event must not let a later one overtake it. After the first
//! event that has to be retried, nothing more of the batch is handled: it and
//! every later event are nakked for immediate redelivery, and the caller waits
//! before it pulls again, which hands them back in stream order.

use anyhow::Result;
use async_nats::jetstream::AckKind;
use futures_util::{Stream, StreamExt as _};

use crate::Skip;

/// One pulled event: what it is called in the log, and how it is acked.
pub(crate) trait Pulled {
    fn subject(&self) -> &str;
    async fn settle(&self, ack: AckKind) -> Result<()>;
}

impl Pulled for async_nats::jetstream::Message {
    fn subject(&self) -> &str {
        &self.subject
    }

    async fn settle(&self, ack: AckKind) -> Result<()> {
        self.ack_with(ack)
            .await
            .map_err(|e| anyhow::anyhow!("acking an event: {e}"))
    }
}

/// Handle each event of `batch` with `handle` and settle it. Returns the
/// error of the first event that has to be retried, once the rest of the
/// batch is nakked behind it.
pub(crate) async fn batch<P: Pulled>(
    batch: impl Stream<Item = Result<P>>,
    mut handle: impl AsyncFnMut(&P) -> Result<(), Skip>,
) -> Result<()> {
    let mut batch = std::pin::pin!(batch);
    let mut failed: Option<anyhow::Error> = None;
    while let Some(event) = batch.next().await {
        let event = event?;
        let ack = if failed.is_some() {
            AckKind::Nak(None)
        } else {
            match handle(&event).await {
                Ok(()) => AckKind::Ack,
                Err(Skip::Unusable(reason)) => {
                    tracing::warn!(subject = event.subject(), %reason, "notify router: dropping an unusable event");
                    AckKind::Term
                }
                Err(Skip::Retry(e)) => {
                    failed = Some(e);
                    AckKind::Nak(None)
                }
            }
        };
        event.settle(ack).await?;
    }
    match failed {
        Some(e) => Err(e.context(
            "an event failed; it and the rest of its batch are redelivered in stream order",
        )),
        None => Ok(()),
    }
}

#[cfg(test)]
mod tests {
    use std::cell::RefCell;

    use super::*;

    struct Fake<'a> {
        name: &'static str,
        settled: &'a RefCell<Vec<(&'static str, String)>>,
    }

    impl Pulled for Fake<'_> {
        fn subject(&self) -> &str {
            self.name
        }

        async fn settle(&self, ack: AckKind) -> Result<()> {
            self.settled
                .borrow_mut()
                .push((self.name, format!("{ack:?}")));
            Ok(())
        }
    }

    /// `assigned` fails, so `unassigned` after it is not handled: both come
    /// back, in order, rather than the unassign overtaking the assign.
    #[tokio::test]
    async fn after_a_retry_the_rest_of_the_batch_is_not_handled_and_comes_back() {
        let settled = RefCell::new(Vec::new());
        let events = ["opened", "assigned", "unassigned", "closed"].map(|name| {
            Ok(Fake {
                name,
                settled: &settled,
            })
        });
        let mut handled = Vec::new();
        let result = batch(futures_util::stream::iter(events), async |e: &Fake<'_>| {
            handled.push(e.name);
            if e.name == "assigned" {
                Err(Skip::Retry(anyhow::anyhow!("the queue is down")))
            } else {
                Ok(())
            }
        })
        .await;
        assert!(result.is_err());
        assert_eq!(handled, ["opened", "assigned"]);
        let nak = format!("{:?}", AckKind::Nak(None));
        assert_eq!(
            settled.into_inner(),
            [
                ("opened", format!("{:?}", AckKind::Ack)),
                ("assigned", nak.clone()),
                ("unassigned", nak.clone()),
                ("closed", nak),
            ]
        );
    }

    /// Control: an unusable event is dropped and the batch goes on.
    #[tokio::test]
    async fn an_unusable_event_is_dropped_and_the_batch_goes_on() {
        let settled = RefCell::new(Vec::new());
        let events = ["bad", "good"].map(|name| {
            Ok(Fake {
                name,
                settled: &settled,
            })
        });
        let result = batch(futures_util::stream::iter(events), async |e: &Fake<'_>| {
            if e.name == "bad" {
                Err(Skip::Unusable("no".to_owned()))
            } else {
                Ok(())
            }
        })
        .await;
        assert!(result.is_ok());
        assert_eq!(
            settled.into_inner(),
            [
                ("bad", format!("{:?}", AckKind::Term)),
                ("good", format!("{:?}", AckKind::Ack)),
            ]
        );
    }
}