//! Backlog replay joined to the live feed.
//!
//! # The ordering that matters
//!
//! Every deploy is a symlink swap and a restart with no connection draining, so
//! every client reconnects and asks to resume from a cursor. Serving that means
//! reading stored messages and then following the live feed, and the order those
//! two things happen in is the whole correctness question.
//!
//! Fetch the backlog first and then subscribe, and every message sent between the
//! two is lost: it is too new to be in the query result and too old to be in the
//! channel. Nothing errors, the client just has a hole exactly where it was busy
//! reconnecting, which is the worst possible moment.
//!
//! So [`ChatStream::open`] subscribes first, then fetches. That inverts the
//! problem into an overlap instead of a gap, and an overlap is fixable: anything
//! the live feed replays that the backlog already covered is dropped by id. The
//! host cannot get this wrong from the outside, because it hands over a fetch
//! closure rather than a finished backlog.

use crate::error::ChatError;
use crate::event::ChatEvent;
use crate::hub::{Hub, Subscription};
use crate::ids::{MessageId, RoomId, UserId};
use crate::message::Message;

/// Backlog replay followed by the live feed, deduplicated across the seam.
pub struct ChatStream {
    backlog: std::vec::IntoIter<Message>,
    live: Subscription,
    /// Highest id served from the backlog. Live messages at or below it are
    /// duplicates from the overlap window and are dropped.
    high_water: Option<MessageId>,
}

impl ChatStream {
    /// Subscribe, then fetch the backlog, then serve one followed by the other.
    ///
    /// `fetch` receives the client's cursor: `Some(id)` to resume after a known
    /// message, `None` for a fresh connection, where the host should return
    /// whatever recent window it wants a new arrival to see. It is called after
    /// the subscription is live, which is the point.
    ///
    /// The host is expected to return messages in ascending id order. If it does
    /// not, the seam still cannot duplicate (the high-water mark is taken from
    /// the maximum, not the last), but the client will see the backlog in
    /// whatever order it was given.
    ///
    /// # Errors
    ///
    /// [`ChatError::ConnectionLimit`] if the hub is at capacity, or whatever
    /// `fetch` fails with.
    pub async fn open<F, Fut>(
        hub: &Hub,
        room: RoomId,
        user: UserId,
        after: Option<MessageId>,
        fetch: F,
    ) -> Result<Self, ChatError>
    where
        F: FnOnce(Option<MessageId>) -> Fut,
        Fut: Future<Output = Result<Vec<Message>, ChatError>>,
    {
        // Subscribe first. Everything published from here on is buffered for us
        // while the fetch runs, so the two sources overlap rather than gap.
        let live = hub.subscribe(room, user)?;

        let backlog = fetch(after).await?;
        let high_water = backlog.iter().map(|m| m.id).max();

        Ok(Self {
            backlog: backlog.into_iter(),
            live,
            high_water,
        })
    }

    /// Next event, or `None` when the room is gone.
    ///
    /// Serves the backlog to exhaustion first, then the live feed.
    pub async fn next(&mut self) -> Option<ChatEvent> {
        if let Some(message) = self.backlog.next() {
            return Some(ChatEvent::Message(message));
        }

        loop {
            let event = self.live.recv().await?;

            // Drop only what the backlog already served. Deletes and purges pass
            // through regardless: a delete that lands during the overlap refers
            // to a message the client has, so suppressing it would leave that
            // message on screen forever.
            if let ChatEvent::Message(m) = &event
                && let Some(high) = self.high_water
                && m.id <= high
            {
                continue;
            }

            return Some(event);
        }
    }

    /// The room being served.
    pub fn room(&self) -> RoomId {
        self.live.room()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hub::HubLimits;
    use crate::ids::Nonce;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use uuid::Uuid;

    fn message(room: RoomId, id: i64) -> Message {
        Message {
            id: MessageId(id),
            room_id: room,
            author_id: UserId(Uuid::nil()),
            body_html: format!("m{id}"),
            created_at: 0,
            nonce: None,
        }
    }

    fn ids(events: &[ChatEvent]) -> Vec<i64> {
        events
            .iter()
            .filter_map(|e| match e {
                ChatEvent::Message(m) => Some(m.id.0),
                _ => None,
            })
            .collect()
    }

    async fn drain(stream: &mut ChatStream, n: usize) -> Vec<ChatEvent> {
        let mut out = Vec::new();
        for _ in 0..n {
            match tokio::time::timeout(std::time::Duration::from_secs(5), stream.next()).await {
                Ok(Some(e)) => out.push(e),
                _ => break,
            }
        }
        out
    }

    #[tokio::test]
    async fn backlog_is_served_before_live() {
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());

        let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
            Ok(vec![message(room, 1), message(room, 2)])
        })
        .await
        .unwrap();

        hub.publish(room, ChatEvent::Message(message(room, 3)));

        assert_eq!(ids(&drain(&mut stream, 3).await), vec![1, 2, 3]);
    }

    #[tokio::test]
    async fn a_message_published_during_the_fetch_is_not_lost() {
        // The failure this exists for: fetch-then-subscribe drops anything sent
        // in the window between the two.
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());
        let hub_for_fetch = hub.clone();

        let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
            // Published while the fetch is in flight, and deliberately absent
            // from what the fetch returns, exactly as a real query would miss a
            // row committed after it ran.
            hub_for_fetch.publish(room, ChatEvent::Message(message(room, 7)));
            Ok(vec![message(room, 5), message(room, 6)])
        })
        .await
        .unwrap();

        assert_eq!(
            ids(&drain(&mut stream, 3).await),
            vec![5, 6, 7],
            "the message sent during the fetch must still arrive"
        );
    }

    #[tokio::test]
    async fn the_overlap_does_not_duplicate() {
        // The other side of subscribe-first: a message can appear in both the
        // backlog and the live feed. It must be served once.
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());
        let hub_for_fetch = hub.clone();

        let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
            hub_for_fetch.publish(room, ChatEvent::Message(message(room, 2)));
            hub_for_fetch.publish(room, ChatEvent::Message(message(room, 3)));
            // The fetch sees both, because they committed before it ran.
            Ok(vec![message(room, 1), message(room, 2), message(room, 3)])
        })
        .await
        .unwrap();

        hub.publish(room, ChatEvent::Message(message(room, 4)));

        assert_eq!(ids(&drain(&mut stream, 4).await), vec![1, 2, 3, 4]);
    }

    #[tokio::test]
    async fn deletes_in_the_overlap_are_not_suppressed() {
        // A delete carries no id that can be compared against the high-water
        // mark, and it refers to a message the client already has. Dropping it
        // would leave a deleted message on screen until reload.
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());
        let hub_for_fetch = hub.clone();

        let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
            hub_for_fetch.publish(room, ChatEvent::Delete { id: MessageId(1) });
            Ok(vec![message(room, 1), message(room, 2)])
        })
        .await
        .unwrap();

        let events = drain(&mut stream, 3).await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ChatEvent::Delete { id: MessageId(1) })),
            "delete during the overlap must pass through: {events:?}"
        );
    }

    #[tokio::test]
    async fn the_cursor_reaches_the_fetch() {
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());
        let seen = Arc::new(AtomicUsize::new(0));
        let seen_clone = seen.clone();

        let _stream = ChatStream::open(
            &hub,
            room,
            UserId(Uuid::nil()),
            Some(MessageId(42)),
            move |after| async move {
                assert_eq!(after, Some(MessageId(42)));
                seen_clone.fetch_add(1, Ordering::SeqCst);
                Ok(vec![])
            },
        )
        .await
        .unwrap();

        assert_eq!(seen.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn an_empty_backlog_goes_straight_to_live() {
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());

        let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
            Ok(vec![])
        })
        .await
        .unwrap();

        hub.publish(room, ChatEvent::Message(message(room, 1)));
        assert_eq!(ids(&drain(&mut stream, 1).await), vec![1]);
    }

    #[tokio::test]
    async fn a_failing_fetch_releases_the_connection_slot() {
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());

        let result = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
            Err(ChatError::MessageEmpty)
        })
        .await;

        assert!(result.is_err());
        assert_eq!(
            hub.connection_count(),
            0,
            "the subscription taken before the fetch must be released when it fails"
        );
        assert_eq!(hub.room_count(), 0);
    }

    #[tokio::test]
    async fn the_nonce_survives_the_seam() {
        // The sender's own optimistic message must arrive carrying its nonce, or
        // the client cannot reconcile and shows it twice.
        let hub = Hub::new(HubLimits::default());
        let room = RoomId(Uuid::new_v4());

        let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
            Ok(vec![])
        })
        .await
        .unwrap();

        let mut m = message(room, 1);
        m.nonce = Nonce::parse("abc");
        hub.publish(room, ChatEvent::Message(m));

        match drain(&mut stream, 1).await.first() {
            Some(ChatEvent::Message(m)) => assert_eq!(m.nonce, Nonce::parse("abc")),
            other => panic!("expected a message carrying its nonce, got {other:?}"),
        }
    }
}