//! The front door.
//!
//! [`Chat`] owns the crate's mutable state (the fan-out hub and the send
//! budgets) so a host holds one thing in application state rather than
//! assembling three and hoping it wires them together the same way twice.

use std::time::{Duration, Instant};

use crate::error::ChatError;
use crate::hub::{Hub, HubLimits};
use crate::ids::{MessageId, Nonce, RoomId, UserId};
use crate::message::Message;
use crate::rate_limit::{RateLimiter, RateLimits};
use crate::room::Room;
use crate::stream::ChatStream;
use crate::traits::ChatAuthz;

/// One send attempt.
#[derive(Debug, Clone)]
pub struct SendRequest<'a> {
    pub room: &'a Room,
    pub author: UserId,
    /// Unvalidated. Trimmed and length-checked on the way through.
    pub body: &'a str,
    /// The sender's optimistic-render nonce, echoed back so their own message
    /// reconciles instead of appearing twice.
    pub nonce: Option<Nonce>,
    /// Injected so the send path is testable without sleeping.
    pub now: Instant,
}

/// Chat state for one process.
///
/// Cheap to clone: the hub is an `Arc` internally and the limiter is shared, so
/// clones talk to the same rooms and the same budgets.
pub struct Chat {
    hub: Hub,
    limiter: RateLimiter,
}

impl Chat {
    pub fn new(hub_limits: HubLimits, rate_limits: RateLimits) -> Self {
        Self {
            hub: Hub::new(hub_limits),
            limiter: RateLimiter::new(rate_limits),
        }
    }

    pub fn hub(&self) -> &Hub {
        &self.hub
    }

    pub fn limiter(&self) -> &RateLimiter {
        &self.limiter
    }

    /// Validate, authorize, store, and fan out one message.
    ///
    /// See [`crate::send`] for why the order of those steps is a security
    /// property rather than a preference.
    pub async fn send<A, F, Fut>(
        &self,
        authz: &A,
        req: SendRequest<'_>,
        store: F,
    ) -> Result<Message, ChatError>
    where
        A: ChatAuthz + ?Sized,
        F: FnOnce(crate::message::MessageBody) -> Fut,
        Fut: Future<Output = Result<Message, ChatError>>,
    {
        crate::send::send(&self.hub, &self.limiter, authz, req, store).await
    }

    /// Open a stream: backlog from `after`, then the live feed.
    ///
    /// Subscribes before fetching. See [`ChatStream`] for why that order is the
    /// whole correctness question.
    pub async fn subscribe<F, Fut>(
        &self,
        room: RoomId,
        user: UserId,
        after: Option<MessageId>,
        fetch: F,
    ) -> Result<ChatStream, ChatError>
    where
        F: FnOnce(Option<MessageId>) -> Fut,
        Fut: Future<Output = Result<Vec<Message>, ChatError>>,
    {
        ChatStream::open(&self.hub, room, user, after, fetch).await
    }

    /// Drop rate-limit buckets nobody has touched recently.
    ///
    /// The host must call this on a timer. Bucket state is keyed by (user, room),
    /// which is unbounded and attacker-influenced, and both consumers run under a
    /// hard 512M cgroup cap. Uses the shortest interval that cannot turn eviction
    /// into a way to skip the queue.
    ///
    /// Returns how many buckets were dropped.
    pub fn sweep(&self, now: Instant) -> usize {
        self.limiter
            .evict_idle(now, self.limiter.recommended_idle_timeout())
    }

    /// How often [`Chat::sweep`] wants to be called.
    pub fn sweep_interval(&self) -> Duration {
        self.limiter.recommended_idle_timeout()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::retention::Retention;
    use crate::room::RoomState;
    use crate::traits::WriteAccess;
    use async_trait::async_trait;
    use uuid::Uuid;

    struct Yes;

    #[async_trait]
    impl ChatAuthz for Yes {
        async fn can_read(&self, _v: Option<UserId>, _r: &Room) -> Result<bool, ChatError> {
            Ok(true)
        }
        async fn can_write(&self, _u: UserId, _r: &Room) -> Result<WriteAccess, ChatError> {
            Ok(WriteAccess::Allow)
        }
        async fn is_moderator(&self, _u: UserId, _r: &Room) -> Result<bool, ChatError> {
            Ok(false)
        }
    }

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

    #[tokio::test]
    async fn send_and_subscribe_share_the_same_rooms() {
        let chat = Chat::new(HubLimits::default(), RateLimits::default());
        let r = room();
        let author = UserId(Uuid::new_v4());

        let mut stream = chat
            .subscribe(r.id, UserId(Uuid::new_v4()), None, |_| async { Ok(vec![]) })
            .await
            .unwrap();

        chat.send(
            &Yes,
            SendRequest {
                room: &r,
                author,
                body: "hi",
                nonce: None,
                now: Instant::now(),
            },
            |body| async move {
                Ok(Message {
                    id: MessageId(1),
                    room_id: r.id,
                    author_id: author,
                    body_html: body.as_str().to_owned(),
                    created_at: 0,
                    nonce: None,
                })
            },
        )
        .await
        .unwrap();

        assert!(
            matches!(
                stream.next().await,
                Some(crate::event::ChatEvent::Message(_))
            ),
            "a message sent through Chat reaches a stream opened through Chat"
        );
    }

    #[test]
    fn sweep_uses_an_interval_that_is_not_a_bypass() {
        let chat = Chat::new(
            HubLimits::default(),
            RateLimits {
                burst: 5,
                sustain_per_min: 60,
            },
        );
        assert_eq!(chat.sweep_interval(), Duration::from_secs(5));

        let t0 = Instant::now();
        chat.limiter()
            .check(UserId(Uuid::nil()), RoomId(Uuid::nil()), t0);
        assert_eq!(chat.sweep(t0), 0, "a fresh bucket is not idle");
        assert_eq!(chat.sweep(t0 + Duration::from_secs(10)), 1);
    }
}