//! The send path.
//!
//! Six things have to happen in a particular order, and the order is a security
//! property rather than a style preference, so it lives here once instead of in
//! each host.
//!
//! 1. **Validate the body.** Free, and rejects the largest class of junk.
//! 2. **Check room state.** Free, and a closed or read-only room ends it.
//! 3. **Take a rate token.** In-memory. Deliberately before authz, so a caller
//!    hammering a room they cannot write to is throttled at memory speed instead
//!    of putting a database query behind every attempt. The cost is that a
//!    refused sender still spends budget, which is the right trade: the budget
//!    exists to protect the room and the box, not to be fair to someone who is
//!    already being refused.
//! 4. **Check authz.** The first step that touches the host's database.
//! 5. **Store.** The host renders through docengine and assigns the id.
//! 6. **Publish.** Only after the row is durable, so nobody sees a message that
//!    a failed insert means will not survive a reconnect.
//!
//! Getting 5 and 6 backwards is the interesting mistake. Publishing first makes
//! chat feel faster and produces a room where a message is visible to everyone
//! present, absent from the backlog, and gone for anyone who reloads.

use crate::chat::SendRequest;
use crate::error::ChatError;
use crate::event::ChatEvent;
use crate::hub::Hub;
use crate::message::{Message, MessageBody};
use crate::rate_limit::{RateDecision, RateLimiter};
use crate::traits::{ChatAuthz, DenyReason, WriteAccess};

/// Why a send was refused, distinguishing the cases a caller must respond to
/// differently.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SendRejection {
    /// The body did not survive validation.
    Invalid(String),
    /// The room is not accepting messages.
    RoomClosed,
    RoomReadOnly,
    /// Authz or rate limiting said no.
    Denied(DenyReason),
}

/// Validate, authorize, store, and fan out one message.
///
/// Reached through [`crate::Chat::send`]; the module docs above explain the
/// ordering it enforces.
///
/// `store` receives the validated body and must render it (through docengine's
/// chat preset, never raw) and insert it, returning the stored [`Message`]. It is
/// called only after every check has passed.
pub(crate) async fn send<A, F, Fut>(
    hub: &Hub,
    limiter: &RateLimiter,
    authz: &A,
    req: SendRequest<'_>,
    store: F,
) -> Result<Message, ChatError>
where
    A: ChatAuthz + ?Sized,
    F: FnOnce(MessageBody) -> Fut,
    Fut: Future<Output = Result<Message, ChatError>>,
{
    let SendRequest {
        room,
        author,
        body: raw_body,
        nonce,
        now,
    } = req;

    // 1. Free.
    let body = MessageBody::parse(raw_body)
        .map_err(|e| ChatError::Rejected(SendRejection::Invalid(e.to_string())))?;

    // 2. Free.
    if !room.state.allows_reads() {
        return Err(ChatError::Rejected(SendRejection::RoomClosed));
    }
    if !room.state.allows_writes() {
        return Err(ChatError::Rejected(SendRejection::RoomReadOnly));
    }

    // 3. In-memory, before anything that queries.
    if let RateDecision::Deny { retry_after } = limiter.check(author, room.id, now) {
        return Err(ChatError::Rejected(SendRejection::Denied(
            DenyReason::RateLimited { retry_after },
        )));
    }

    // 4. First database hit.
    if let WriteAccess::Deny(reason) = authz.can_write(author, room).await? {
        return Err(ChatError::Rejected(SendRejection::Denied(reason)));
    }

    // 5. Durable before visible.
    let mut stored = store(body).await?;
    stored.nonce = nonce;

    // 6. Fan out. A room with nobody in it is normal and not a failure.
    hub.publish(room.id, ChatEvent::Message(stored.clone()));

    Ok(stored)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hub::HubLimits;
    use crate::ids::{MessageId, Nonce, RoomId, UserId};
    use crate::message::MAX_MESSAGE_LEN;
    use crate::rate_limit::RateLimits;
    use crate::retention::Retention;
    use crate::room::{Room, RoomState};
    use async_trait::async_trait;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Instant;
    use uuid::Uuid;

    struct Authz {
        decision: WriteAccess,
        calls: Arc<AtomicUsize>,
    }

    #[async_trait]
    impl ChatAuthz for Authz {
        async fn can_read(&self, _v: Option<UserId>, room: &Room) -> Result<bool, ChatError> {
            Ok(room.state.allows_reads())
        }
        async fn can_write(&self, _u: UserId, _r: &Room) -> Result<WriteAccess, ChatError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Ok(self.decision.clone())
        }
        async fn is_moderator(&self, _u: UserId, _r: &Room) -> Result<bool, ChatError> {
            Ok(false)
        }
    }

    fn authz(decision: WriteAccess) -> (Authz, Arc<AtomicUsize>) {
        let calls = Arc::new(AtomicUsize::new(0));
        (
            Authz {
                decision,
                calls: calls.clone(),
            },
            calls,
        )
    }

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

    fn stored(room: RoomId, author: UserId, body: &MessageBody) -> Message {
        Message {
            id: MessageId(1),
            room_id: room,
            author_id: author,
            body_html: format!("<p>{}</p>", body.as_str()),
            created_at: 0,
            nonce: None,
        }
    }

    fn parts() -> (Hub, RateLimiter) {
        (
            Hub::new(HubLimits::default()),
            RateLimiter::new(RateLimits::default()),
        )
    }

    fn req<'a>(room: &'a Room, author: UserId, body: &'a str) -> SendRequest<'a> {
        SendRequest {
            room,
            author,
            body,
            nonce: None,
            now: Instant::now(),
        }
    }

    #[tokio::test]
    async fn a_good_message_is_stored_then_published() {
        let (hub, rl) = parts();
        let (a, _) = authz(WriteAccess::Allow);
        let r = room(RoomState::Open);
        let author = UserId(Uuid::new_v4());
        let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();

        let out = send(
            &hub,
            &rl,
            &a,
            SendRequest {
                nonce: Nonce::parse("n1"),
                ..req(&r, author, "  hello  ")
            },
            |body| async move { Ok(stored(r.id, author, &body)) },
        )
        .await
        .unwrap();

        assert_eq!(
            out.body_html, "<p>hello</p>",
            "body was trimmed and rendered"
        );
        assert_eq!(out.nonce, Nonce::parse("n1"));

        match listener.recv().await {
            Some(ChatEvent::Message(m)) => {
                assert_eq!(m.id, MessageId(1));
                assert_eq!(m.nonce, Nonce::parse("n1"), "sender can reconcile");
            }
            other => panic!("expected the message on the room, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn nothing_is_published_when_the_store_fails() {
        // The ordering this protects: a message visible to everyone present but
        // absent from the backlog and gone on reload.
        let (hub, rl) = parts();
        let (a, _) = authz(WriteAccess::Allow);
        let r = room(RoomState::Open);
        let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();

        let result = send(
            &hub,
            &rl,
            &a,
            req(&r, UserId(Uuid::new_v4()), "hi"),
            |_| async { Err(ChatError::host(std::io::Error::other("db down"))) },
        )
        .await;

        assert!(result.is_err());
        // Probe the room. If a phantom message had been published it would be
        // ahead of this in the queue.
        assert_eq!(hub.publish(r.id, ChatEvent::Gap), 1, "listener still live");
        assert!(
            matches!(listener.recv().await, Some(ChatEvent::Gap)),
            "a failed store must publish nothing"
        );
    }

    #[tokio::test]
    async fn an_empty_body_never_reaches_authz() {
        let (hub, rl) = parts();
        let (a, calls) = authz(WriteAccess::Allow);
        let r = room(RoomState::Open);

        let result = send(
            &hub,
            &rl,
            &a,
            req(&r, UserId(Uuid::new_v4()), "   "),
            |_| async { panic!("store must not run") },
        )
        .await;

        assert!(matches!(
            result,
            Err(ChatError::Rejected(SendRejection::Invalid(_)))
        ));
        assert_eq!(calls.load(Ordering::SeqCst), 0, "no query for a blank body");
    }

    #[tokio::test]
    async fn an_overlong_body_is_refused() {
        let (hub, rl) = parts();
        let (a, _) = authz(WriteAccess::Allow);
        let r = room(RoomState::Open);
        let long = "x".repeat(MAX_MESSAGE_LEN + 1);

        let result = send(
            &hub,
            &rl,
            &a,
            req(&r, UserId(Uuid::new_v4()), &long),
            |_| async { panic!("store must not run") },
        )
        .await;

        assert!(matches!(
            result,
            Err(ChatError::Rejected(SendRejection::Invalid(_)))
        ));
    }

    #[tokio::test]
    async fn a_read_only_room_refuses_before_querying() {
        let (hub, rl) = parts();
        let (a, calls) = authz(WriteAccess::Allow);
        let r = room(RoomState::ReadOnly);

        let result = send(
            &hub,
            &rl,
            &a,
            req(&r, UserId(Uuid::new_v4()), "hi"),
            |_| async { panic!("store must not run") },
        )
        .await;

        assert!(matches!(
            result,
            Err(ChatError::Rejected(SendRejection::RoomReadOnly))
        ));
        assert_eq!(calls.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn a_closed_room_refuses_as_closed_not_read_only() {
        let (hub, rl) = parts();
        let (a, _) = authz(WriteAccess::Allow);
        let r = room(RoomState::Closed);

        let result = send(
            &hub,
            &rl,
            &a,
            req(&r, UserId(Uuid::new_v4()), "hi"),
            |_| async { panic!("store must not run") },
        )
        .await;

        assert!(matches!(
            result,
            Err(ChatError::Rejected(SendRejection::RoomClosed))
        ));
    }

    #[tokio::test]
    async fn a_denied_sender_gets_the_specific_reason() {
        let (hub, rl) = parts();
        let (a, _) = authz(WriteAccess::Deny(DenyReason::Muted));
        let r = room(RoomState::Open);

        let result = send(
            &hub,
            &rl,
            &a,
            req(&r, UserId(Uuid::new_v4()), "hi"),
            |_| async { panic!("store must not run") },
        )
        .await;

        assert!(matches!(
            result,
            Err(ChatError::Rejected(SendRejection::Denied(
                DenyReason::Muted
            )))
        ));
    }

    #[tokio::test]
    async fn the_rate_limit_bites_before_authz_is_queried() {
        // A caller hammering a room is throttled in memory rather than putting a
        // query behind every attempt.
        let hub = Hub::new(HubLimits::default());
        let rl = RateLimiter::new(RateLimits {
            burst: 2,
            sustain_per_min: 60,
        });
        let (a, calls) = authz(WriteAccess::Allow);
        let r = room(RoomState::Open);
        let author = UserId(Uuid::new_v4());
        let t0 = Instant::now();

        let at = |body| SendRequest {
            room: &r,
            author,
            body,
            nonce: None,
            now: t0,
        };

        for _ in 0..2 {
            send(&hub, &rl, &a, at("hi"), |body| async move {
                Ok(stored(r.id, author, &body))
            })
            .await
            .unwrap();
        }
        assert_eq!(calls.load(Ordering::SeqCst), 2);

        for _ in 0..10 {
            let result = send(&hub, &rl, &a, at("hi"), |_| async {
                panic!("store must not run")
            })
            .await;
            assert!(matches!(
                result,
                Err(ChatError::Rejected(SendRejection::Denied(
                    DenyReason::RateLimited { .. }
                )))
            ));
        }
        assert_eq!(
            calls.load(Ordering::SeqCst),
            2,
            "throttled attempts must not reach the database"
        );
    }
}