//! The moderation path.
//!
//! Same discipline as [`crate::send`], for the same reason: the order is a
//! correctness property, so it lives here once rather than in each host.
//! **Mutate durably, then publish.** Broadcasting a removal the database
//! refused produces a room where a message is gone for everyone present and
//! back for anyone who reloads, which is the deletion mirror of the phantom
//! message the send path is ordered to avoid.
//!
//! # Why a ban is one event
//!
//! A ban removes every message its target wrote in the retention window, and
//! that is a burst with no useful upper bound: the whole point of banning
//! someone is usually that they said a great many things. Fanning it out as one
//! [`ChatEvent::Delete`] per message would push that burst through every open
//! connection in the room at once, and the room buffer is finite, so past a few
//! hundred messages the deletes would overflow it and every listener would take
//! a [`ChatEvent::Gap`] and re-fetch the backlog. A moderation action that
//! reliably triggers a stampede is one moderators learn not to use.
//!
//! So the removal is a single statement keyed by (room, author) and a single
//! [`ChatEvent::Purge`], which the client applies by dropping everything it
//! holds from that author. Cost is constant in the number of messages.

use crate::error::ChatError;
use crate::event::ChatEvent;
use crate::hub::Hub;
use crate::ids::{MessageId, UserId};
use crate::room::Room;
use crate::traits::ChatModeration;

/// Remove one message, then tell the room.
///
/// The host's [`ChatModeration::delete_message`] is responsible for proving
/// `actor` is entitled to this message, which is why the entitlement check is
/// not duplicated here: it needs the row, and only the host can read it.
pub(crate) async fn delete_message<M>(
    hub: &Hub,
    moderation: &M,
    actor: UserId,
    room: &Room,
    message: MessageId,
) -> Result<(), ChatError>
where
    M: ChatModeration + ?Sized,
{
    moderation.delete_message(actor, room, message).await?;
    hub.publish(room.id, ChatEvent::Delete { id: message });
    Ok(())
}

/// Remove every message by `target` in the room, then tell the room once.
///
/// Returns how many rows went.
pub(crate) async fn purge_user<M>(
    hub: &Hub,
    moderation: &M,
    actor: UserId,
    room: &Room,
    target: UserId,
) -> Result<u64, ChatError>
where
    M: ChatModeration + ?Sized,
{
    let removed = moderation.purge_user(actor, room, target).await?;

    // Broadcast even when nothing was removed. A connected client can be
    // holding messages the retention sweep already deleted server-side, so
    // "zero rows" does not mean "nothing on screen", and the event is one frame
    // that clients apply idempotently.
    hub.publish(room.id, ChatEvent::Purge { author_id: target });

    Ok(removed)
}

/// Ban `target` and remove their backlog in one action.
///
/// Returns how many messages the ban removed.
///
/// The ban lands first. If the purge then fails, the caller gets the error and
/// the ban still stands, which is the right way round: a banned user with
/// messages still up is a cleanup job, while a purged user who can still type
/// is the incident continuing.
pub(crate) async fn ban_user<M>(
    hub: &Hub,
    moderation: &M,
    actor: UserId,
    room: &Room,
    target: UserId,
    reason: Option<&str>,
) -> Result<u64, ChatError>
where
    M: ChatModeration + ?Sized,
{
    moderation.ban_user(actor, room, target, reason).await?;
    purge_user(hub, moderation, actor, room, target).await
}

/// Stop `target` from sending for `duration`.
///
/// Publishes nothing, deliberately. A timeout removes no content, so there is
/// nothing for a connected client to re-render, and reads are explicitly
/// unaffected. Announcing it to the room would turn a quiet moderation action
/// into a public one and hand every listener an event whose only content is
/// somebody else's punishment. The person timed out finds out when their next
/// send is refused with [`DenyReason::Muted`](crate::DenyReason::Muted).
pub(crate) async fn timeout_user<M>(
    moderation: &M,
    actor: UserId,
    room: &Room,
    target: UserId,
    duration: std::time::Duration,
) -> Result<(), ChatError>
where
    M: ChatModeration + ?Sized,
{
    moderation.timeout_user(actor, room, target, duration).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hub::HubLimits;
    use crate::ids::RoomId;
    use crate::retention::Retention;
    use crate::room::RoomState;
    use async_trait::async_trait;
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;
    use uuid::Uuid;

    /// Records what the host was asked to do, and can be told to fail.
    #[derive(Default)]
    struct Recorder {
        calls: Mutex<Vec<String>>,
        fail_ban: AtomicBool,
        fail_purge: AtomicBool,
        fail_delete: AtomicBool,
        purged_rows: u64,
    }

    impl Recorder {
        fn calls(&self) -> Vec<String> {
            self.calls.lock().unwrap().clone()
        }
        fn note(&self, what: &str) {
            self.calls.lock().unwrap().push(what.to_owned());
        }
    }

    fn boom() -> ChatError {
        ChatError::host(std::io::Error::other("db down"))
    }

    #[async_trait]
    impl ChatModeration for Recorder {
        async fn delete_message(
            &self,
            _actor: UserId,
            _room: &Room,
            message: MessageId,
        ) -> Result<(), ChatError> {
            self.note(&format!("delete {message}"));
            if self.fail_delete.load(Ordering::SeqCst) {
                return Err(boom());
            }
            Ok(())
        }

        async fn purge_user(
            &self,
            _actor: UserId,
            _room: &Room,
            _target: UserId,
        ) -> Result<u64, ChatError> {
            self.note("purge");
            if self.fail_purge.load(Ordering::SeqCst) {
                return Err(boom());
            }
            Ok(self.purged_rows)
        }

        async fn timeout_user(
            &self,
            _actor: UserId,
            _room: &Room,
            _target: UserId,
            _duration: Duration,
        ) -> Result<(), ChatError> {
            self.note("timeout");
            Ok(())
        }

        async fn ban_user(
            &self,
            _actor: UserId,
            _room: &Room,
            _target: UserId,
            _reason: Option<&str>,
        ) -> Result<(), ChatError> {
            self.note("ban");
            if self.fail_ban.load(Ordering::SeqCst) {
                return Err(boom());
            }
            Ok(())
        }

        async fn log_action(
            &self,
            _actor: UserId,
            _room_id: RoomId,
            _action: &str,
            _detail: Option<&str>,
        ) -> Result<(), ChatError> {
            Ok(())
        }
    }

    fn room() -> Room {
        Room {
            id: RoomId(Uuid::new_v4()),
            state: RoomState::Open,
            retention: Retention::forum_default(),
        }
    }

    fn user() -> UserId {
        UserId(Uuid::new_v4())
    }

    /// Everything already queued on the room, without blocking.
    ///
    /// Polled rather than awaited because half these tests assert that *no*
    /// event was published, and awaiting for that would mean picking a timeout
    /// and trading a fast suite against a flaky one. Every event under test is
    /// published before the assertion runs, so anything not queued by now is
    /// genuinely absent.
    fn drained(sub: &mut crate::hub::Subscription) -> Vec<ChatEvent> {
        use std::task::{Context, Poll, Waker};

        let mut out = Vec::new();
        let mut cx = Context::from_waker(Waker::noop());
        loop {
            let mut recv = Box::pin(sub.recv());
            match recv.as_mut().poll(&mut cx) {
                Poll::Ready(Some(event)) => out.push(event),
                Poll::Ready(None) | Poll::Pending => return out,
            }
        }
    }

    #[tokio::test]
    async fn a_delete_reaches_the_room() {
        let hub = Hub::new(HubLimits::default());
        let r = room();
        let mut listener = hub.subscribe(r.id, user()).unwrap();

        delete_message(&hub, &Recorder::default(), user(), &r, MessageId(7))
            .await
            .unwrap();

        assert_eq!(
            drained(&mut listener),
            vec![ChatEvent::Delete { id: MessageId(7) }]
        );
    }

    #[tokio::test]
    async fn a_failed_delete_publishes_nothing() {
        // The mirror of the phantom-message failure: a message gone for
        // everyone present and back on reload.
        let hub = Hub::new(HubLimits::default());
        let r = room();
        let mut listener = hub.subscribe(r.id, user()).unwrap();

        let host = Recorder::default();
        host.fail_delete.store(true, Ordering::SeqCst);

        assert!(
            delete_message(&hub, &host, user(), &r, MessageId(7))
                .await
                .is_err()
        );
        assert!(
            drained(&mut listener).is_empty(),
            "a refused delete must not be broadcast"
        );
    }

    #[tokio::test]
    async fn a_ban_purges_and_emits_exactly_one_event() {
        // The property the whole module exists for: constant event cost
        // regardless of how much the banned user wrote.
        let hub = Hub::new(HubLimits::default());
        let r = room();
        let target = user();
        let mut listener = hub.subscribe(r.id, user()).unwrap();

        let host = Recorder {
            purged_rows: 4_000,
            ..Default::default()
        };

        let removed = ban_user(&hub, &host, user(), &r, target, Some("spam"))
            .await
            .unwrap();

        assert_eq!(removed, 4_000);
        assert_eq!(host.calls(), vec!["ban", "purge"], "ban lands before purge");
        assert_eq!(
            drained(&mut listener),
            vec![ChatEvent::Purge { author_id: target }],
            "4000 messages removed, one event"
        );
    }

    #[tokio::test]
    async fn a_failed_ban_never_purges() {
        let hub = Hub::new(HubLimits::default());
        let r = room();
        let mut listener = hub.subscribe(r.id, user()).unwrap();

        let host = Recorder::default();
        host.fail_ban.store(true, Ordering::SeqCst);

        assert!(
            ban_user(&hub, &host, user(), &r, user(), None)
                .await
                .is_err()
        );
        assert_eq!(host.calls(), vec!["ban"], "the purge must not have run");
        assert!(drained(&mut listener).is_empty());
    }

    #[tokio::test]
    async fn a_ban_whose_purge_fails_still_reports_the_error() {
        let hub = Hub::new(HubLimits::default());
        let r = room();
        let mut listener = hub.subscribe(r.id, user()).unwrap();

        let host = Recorder::default();
        host.fail_purge.store(true, Ordering::SeqCst);

        assert!(
            ban_user(&hub, &host, user(), &r, user(), None)
                .await
                .is_err(),
            "the caller must learn the backlog is still up"
        );
        assert_eq!(host.calls(), vec!["ban", "purge"]);
        assert!(
            drained(&mut listener).is_empty(),
            "nothing was removed, so nothing is announced"
        );
    }

    #[tokio::test]
    async fn a_purge_that_removed_nothing_is_still_announced() {
        // A client can hold messages the retention sweep already deleted, so
        // zero rows does not mean nothing on screen.
        let hub = Hub::new(HubLimits::default());
        let r = room();
        let target = user();
        let mut listener = hub.subscribe(r.id, user()).unwrap();

        let removed = purge_user(&hub, &Recorder::default(), user(), &r, target)
            .await
            .unwrap();

        assert_eq!(removed, 0);
        assert_eq!(
            drained(&mut listener),
            vec![ChatEvent::Purge { author_id: target }]
        );
    }

    #[tokio::test]
    async fn a_timeout_is_not_announced_to_the_room() {
        let hub = Hub::new(HubLimits::default());
        let r = room();
        let mut listener = hub.subscribe(r.id, user()).unwrap();

        let host = Recorder::default();
        timeout_user(&host, user(), &r, user(), Duration::from_mins(5))
            .await
            .unwrap();

        assert_eq!(host.calls(), vec!["timeout"]);
        assert!(
            drained(&mut listener).is_empty(),
            "a timeout removes no content, so there is nothing to re-render"
        );
    }
}